Skip to content

Commit daa31f6

Browse files
hwjiangkaihwjiangkai
committed
remove vanus sdk in go mod file
Signed-off-by: jyjiangkai <[email protected]>
1 parent 14cc0d1 commit daa31f6

File tree

11 files changed

+98
-130
lines changed

11 files changed

+98
-130
lines changed

‎observability/tracing/exporter/event_trace_exporter.go renamed to ‎client/pkg/exporter/event_trace_exporter.go

+58-32
Original file line numberDiff line numberDiff line change
@@ -15,34 +15,54 @@
1515
package exporter
1616

1717
import (
18+
// standard libraries.
1819
"context"
20+
"os"
1921

22+
// third-party libraries.
2023
v2 "github.com/cloudevents/sdk-go/v2"
2124
"github.com/google/uuid"
22-
vanussdk "github.com/vanus-labs/sdk/golang"
23-
"github.com/vanus-labs/vanus/observability/log"
24-
2525
"go.opentelemetry.io/otel/attribute"
2626
tracesdk "go.opentelemetry.io/otel/sdk/trace"
27+
"google.golang.org/grpc/credentials/insecure"
28+
29+
// first-party libraries.
30+
"github.com/vanus-labs/vanus/observability/log"
31+
"github.com/vanus-labs/vanus/pkg/cluster"
32+
"github.com/vanus-labs/vanus/proto/pkg/cloudevents"
33+
"github.com/vanus-labs/vanus/proto/pkg/codec"
34+
35+
// this project.
36+
"github.com/vanus-labs/vanus/client"
37+
"github.com/vanus-labs/vanus/client/pkg/api"
2738
)
2839

40+
func GetExporter(endpoints []string, eventbus string) tracesdk.SpanExporter {
41+
spanExporter, err := New(context.Background(), WithEndpoints(endpoints), WithEventbus(eventbus))
42+
if err != nil {
43+
log.Error().Err(err).Msg("new span exporter failed")
44+
os.Exit(-1)
45+
}
46+
return spanExporter
47+
}
48+
2949
type Option func(*Options)
3050

3151
type Options struct {
32-
Endpoints string
52+
Endpoints []string
3353
Eventbus string
3454
}
3555

3656
func defaultOptions() *Options {
3757
return &Options{
38-
Endpoints: "127.0.0.1:8080",
58+
Endpoints: []string{},
3959
Eventbus: "event-tracing",
4060
}
4161
}
4262

43-
func WithEndpoint(endpoint string) Option {
63+
func WithEndpoints(endpoints []string) Option {
4464
return func(options *Options) {
45-
options.Endpoints = endpoint
65+
options.Endpoints = endpoints
4666
}
4767
}
4868

@@ -54,9 +74,8 @@ func WithEventbus(eventbus string) Option {
5474

5575
// Exporter exports trace data in the OTLP wire format.
5676
type Exporter struct {
57-
endpoints string
58-
client vanussdk.Client
59-
publisher vanussdk.Publisher
77+
endpoints []string
78+
writer api.BusWriter
6079
}
6180

6281
var _ tracesdk.SpanExporter = (*Exporter)(nil)
@@ -67,53 +86,58 @@ func New(ctx context.Context, opts ...Option) (*Exporter, error) {
6786
apply(defaultOpts)
6887
}
6988

70-
clientOpts := &vanussdk.ClientOptions{
71-
Endpoint: defaultOpts.Endpoints,
72-
Token: "admin",
89+
ctrl := cluster.NewClusterController(defaultOpts.Endpoints, insecure.NewCredentials())
90+
if err := ctrl.WaitForControllerReady(true); err != nil {
91+
log.Error(ctx).Err(err).Msg("wait for controller ready timeout")
92+
return nil, err
7393
}
74-
75-
c, err := vanussdk.Connect(clientOpts)
94+
eventbus, err := ctrl.EventbusService().GetEventbusByName(ctx, "default", defaultOpts.Eventbus)
7695
if err != nil {
77-
panic("failed to connect to Vanus cluster, error: " + err.Error())
96+
log.Error(ctx).Err(err).Str("eventbus", defaultOpts.Eventbus).Msg("failed to get eventbus")
97+
return nil, err
7898
}
7999

80-
ebOpt := vanussdk.WithEventbus("default", defaultOpts.Eventbus)
100+
c := client.Connect(defaultOpts.Endpoints)
101+
bus := c.Eventbus(ctx, api.WithName(defaultOpts.Eventbus), api.WithID(eventbus.Id))
81102
exporter := &Exporter{
82103
endpoints: defaultOpts.Endpoints,
83-
client: c,
84-
publisher: c.Publisher(ebOpt),
85-
}
86-
_, err = c.Controller().Eventbus().Get(ctx, ebOpt)
87-
if err != nil {
88-
panic("failed to get tracing eventbus, error: " + err.Error())
104+
writer: bus.Writer(),
89105
}
90106
return exporter, nil
91107
}
92108

93109
// ExportSpans exports a batch of spans.
94110
func (e *Exporter) ExportSpans(ctx context.Context, ss []tracesdk.ReadOnlySpan) error {
95-
es := make([]*v2.Event, 0)
111+
ces := make([]*cloudevents.CloudEvent, 0)
96112
for _, span := range ss {
97-
if span.Name() != "EventTracing" {
113+
event := newEvent(span)
114+
if event.Type() != "event-tracing" {
98115
continue
99116
}
100-
event := newEvent(span)
101-
es = append(es, &event)
117+
eventpb, err := codec.ToProto(&event)
118+
if err != nil {
119+
log.Error(ctx).Err(err).Any("event", event).Msg("failed to proto event")
120+
return nil
121+
}
122+
ces = append(ces, eventpb)
102123
}
103124

104-
if len(es) == 0 {
125+
if len(ces) == 0 {
105126
return nil
106127
}
107128

108-
err := e.publisher.Publish(ctx, es...)
129+
ceBatch := &cloudevents.CloudEventBatch{
130+
Events: ces,
131+
}
132+
_, err := e.writer.Append(ctx, ceBatch)
109133
if err != nil {
110-
log.Error(ctx).Err(err).Msg("failed to publish events to tracing eventbus")
134+
log.Error(ctx).Err(err).Msg("failed to append events to tracing eventbus")
111135
return nil
112136
}
113137
return nil
114138
}
115139

116-
// Shutdown flushes all exports and closes all connections to the receiving endpoint.
140+
// Shutdown
117141
func (e *Exporter) Shutdown(ctx context.Context) error {
118142
return nil
119143
}
@@ -122,7 +146,6 @@ func newEvent(span tracesdk.ReadOnlySpan) v2.Event {
122146
event := v2.NewEvent()
123147
event.SetID(uuid.New().String())
124148
event.SetSource(span.Name())
125-
event.SetType(span.SpanKind().String())
126149
data := make(map[string]interface{})
127150
data["name"] = span.Name()
128151
data["trace_id"] = span.SpanContext().TraceID().String()
@@ -136,6 +159,9 @@ func newEvent(span tracesdk.ReadOnlySpan) v2.Event {
136159
data[string(attr.Key)] = attr.Value.AsInt64()
137160
} else if attr.Value.Type() == attribute.STRING {
138161
data[string(attr.Key)] = attr.Value.AsString()
162+
if string(attr.Key) == "type" && attr.Value.AsString() == "event-tracing" {
163+
event.SetType("event-tracing")
164+
}
139165
}
140166
}
141167
data["events"] = span.Events()

‎cmd/gateway/main.go

+2-1
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import (
1818
"flag"
1919
"os"
2020

21+
"github.com/vanus-labs/vanus/client/pkg/exporter"
2122
"github.com/vanus-labs/vanus/observability"
2223
"github.com/vanus-labs/vanus/observability/log"
2324
"github.com/vanus-labs/vanus/observability/metrics"
@@ -39,7 +40,7 @@ func main() {
3940

4041
ctx := signal.SetupSignalContext()
4142
cfg.Observability.T.ServerName = "Vanus Gateway"
42-
_ = observability.Initialize(ctx, cfg.Observability, metrics.GetGatewayMetrics)
43+
_ = observability.Initialize(ctx, cfg.Observability, metrics.GetGatewayMetrics, exporter.GetExporter)
4344

4445
ga := gateway.NewGateway(*cfg)
4546
if err = ga.Start(ctx); err != nil {

‎cmd/trigger/main.go

+2-1
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323

2424
"google.golang.org/grpc"
2525

26+
"github.com/vanus-labs/vanus/client/pkg/exporter"
2627
"github.com/vanus-labs/vanus/observability"
2728
"github.com/vanus-labs/vanus/observability/log"
2829
"github.com/vanus-labs/vanus/observability/metrics"
@@ -50,7 +51,7 @@ func main() {
5051
}
5152
ctx := signal.SetupSignalContext()
5253
cfg.Observability.T.ServerName = "Vanus Trigger"
53-
_ = observability.Initialize(ctx, cfg.Observability, metrics.GetTriggerMetrics)
54+
_ = observability.Initialize(ctx, cfg.Observability, metrics.GetTriggerMetrics, exporter.GetExporter)
5455
var opts []grpc.ServerOption
5556
grpcServer := grpc.NewServer(opts...)
5657
srv := trigger.NewTriggerServer(*cfg)

‎go.mod

+1-3
Original file line numberDiff line numberDiff line change
@@ -48,8 +48,6 @@ 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-
5351
require (
5452
cloud.google.com/go/compute v1.19.0 // indirect
5553
cloud.google.com/go/compute/metadata v0.2.3 // indirect
@@ -129,4 +127,4 @@ replace (
129127
github.com/vanus-labs/vanus/raft => ./raft
130128
)
131129

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

‎go.sum

-2
Original file line numberDiff line numberDiff line change
@@ -325,8 +325,6 @@ 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=
330328
github.com/vigneshuvi/GoDateFormat v0.0.0-20210204121036-67364dc23c79 h1:37VzBuFO88QQnCEu+G41v9IqgJNBXR+4vR9vGwVqJ00=
331329
github.com/vigneshuvi/GoDateFormat v0.0.0-20210204121036-67364dc23c79/go.mod h1:190gFTWxRNREiiPal7zWZlNrwFSpv3BxDmOfgYqoYCY=
332330
github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=

‎internal/gateway/gateway.go

+6-2
Original file line numberDiff line numberDiff line change
@@ -106,8 +106,12 @@ func (ga *ceGateway) receive(ctx context.Context, event v2.Event) (re *v2.Event,
106106
ctx, span := ga.tracer.Start(ctx, "receive")
107107
defer span.End()
108108

109-
event.SetExtension("traceid", span.SpanContext().TraceID().String())
110-
span.SetName("EventTracing")
109+
traceID := span.SpanContext().TraceID().String()
110+
spanID := span.SpanContext().SpanID().String()
111+
traceFlags := span.SpanContext().TraceFlags().String()
112+
event.SetExtension("traceparent", fmt.Sprintf("00-%s-%s-%s", traceID, spanID, traceFlags))
113+
span.SetName(event.ID())
114+
span.SetAttributes(attribute.String("type", "event-tracing"))
111115
span.SetAttributes(attribute.String("event_id", event.ID()))
112116
span.AddEvent("received from source", trace.WithTimestamp(time.Now()))
113117

‎internal/trigger/reader/reader.go

+7-4
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import (
1919
"context"
2020
"encoding/binary"
2121
stderr "errors"
22+
"strings"
2223
"sync"
2324
"time"
2425

@@ -206,7 +207,8 @@ func (elReader *eventlogReader) loop(ctx context.Context, lr api.BusReader) erro
206207
}
207208
for i := range events {
208209
_, span := elReader.newSpan(ctx, *events[i])
209-
span.SetName("EventTracing")
210+
span.SetName(events[i].ID())
211+
span.SetAttributes(attribute.String("type", "event-tracing"))
210212
span.SetAttributes(attribute.String("event_id", events[i].ID()))
211213
span.AddEvent("read from eventbus", trace.WithTimestamp(time.Now()))
212214
span.SetAttributes(attribute.String("eventbus_id", elReader.config.EventbusID.String()))
@@ -240,12 +242,13 @@ func (elReader *eventlogReader) newSpan(ctx context.Context, event ce.Event) (co
240242
if event.Extensions() == nil {
241243
return elReader.config.Tracer.Start(ctx, event.ID())
242244
}
243-
if _, ok := event.Extensions()["traceid"]; !ok {
245+
if _, ok := event.Extensions()["traceparent"]; !ok {
244246
return elReader.config.Tracer.Start(ctx, event.ID())
245247
}
246-
traceid, _ := trace.TraceIDFromHex(event.Extensions()["traceid"].(string))
248+
tps := strings.Split(event.Extensions()["traceparent"].(string), "-")
249+
traceID, _ := trace.TraceIDFromHex(tps[1])
247250
_ctx := trace.ContextWithSpanContext(ctx, trace.NewSpanContext(trace.SpanContextConfig{
248-
TraceID: traceid,
251+
TraceID: traceID,
249252
}))
250253
return elReader.config.Tracer.Start(_ctx, event.ID())
251254
}

‎observability/go.mod

+1-16
Original file line numberDiff line numberDiff line change
@@ -3,11 +3,8 @@ module github.com/vanus-labs/vanus/observability
33
go 1.19
44

55
require (
6-
github.com/cloudevents/sdk-go/v2 v2.14.0
7-
github.com/google/uuid v1.3.0
86
github.com/prometheus/client_golang v1.15.0
97
github.com/rs/zerolog v1.29.1
10-
github.com/vanus-labs/sdk/golang v0.4.7
118
go.opentelemetry.io/otel v1.14.0
129
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.14.0
1310
go.opentelemetry.io/otel/sdk v1.14.0
@@ -21,34 +18,22 @@ require (
2118
github.com/cespare/xxhash/v2 v2.2.0 // indirect
2219
github.com/go-logr/logr v1.2.3 // indirect
2320
github.com/go-logr/stdr v1.2.2 // indirect
24-
github.com/golang/mock v1.6.0 // indirect
2521
github.com/golang/protobuf v1.5.3 // indirect
2622
github.com/grpc-ecosystem/grpc-gateway/v2 v2.11.3 // indirect
27-
github.com/json-iterator/go v1.1.12 // indirect
2823
github.com/mattn/go-colorable v0.1.13 // indirect
2924
github.com/mattn/go-isatty v0.0.17 // indirect
3025
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
31-
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
32-
github.com/modern-go/reflect2 v1.0.2 // indirect
33-
github.com/pkg/errors v0.9.1 // indirect
3426
github.com/prometheus/client_model v0.3.0 // indirect
3527
github.com/prometheus/common v0.42.0 // indirect
3628
github.com/prometheus/procfs v0.9.0 // indirect
37-
github.com/vanus-labs/vanus/pkg v0.0.0-20230330121558-f90ea26ae55b // indirect
38-
github.com/vanus-labs/vanus/proto v0.7.1 // indirect
3929
go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.14.0 // indirect
4030
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.14.0 // indirect
4131
go.opentelemetry.io/proto/otlp v0.19.0 // indirect
42-
go.uber.org/atomic v1.4.0 // indirect
43-
go.uber.org/multierr v1.1.0 // indirect
44-
go.uber.org/zap v1.10.0 // indirect
4532
golang.org/x/net v0.9.0 // indirect
46-
golang.org/x/oauth2 v0.5.0 // indirect
4733
golang.org/x/sys v0.7.0 // indirect
4834
golang.org/x/text v0.9.0 // indirect
49-
google.golang.org/appengine v1.6.7 // indirect
5035
google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
5136
google.golang.org/protobuf v1.30.0 // indirect
5237
)
5338

54-
// replace github.com/vanus-labs/vanus => ../FORBIDDEN_DEPENDENCY
39+
replace github.com/vanus-labs/vanus => ../FORBIDDEN_DEPENDENCY

0 commit comments

Comments
 (0)