Skip to content

Commit 58b04d8

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

File tree

11 files changed

+87
-127
lines changed

11 files changed

+87
-127
lines changed

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

Lines changed: 50 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -16,33 +16,48 @@ package exporter
1616

1717
import (
1818
"context"
19+
"os"
1920

2021
v2 "github.com/cloudevents/sdk-go/v2"
2122
"github.com/google/uuid"
22-
vanussdk "github.com/vanus-labs/sdk/golang"
23+
"github.com/vanus-labs/vanus/client"
24+
"github.com/vanus-labs/vanus/client/pkg/api"
2325
"github.com/vanus-labs/vanus/observability/log"
26+
"github.com/vanus-labs/vanus/pkg/cluster"
27+
"github.com/vanus-labs/vanus/proto/pkg/cloudevents"
28+
"github.com/vanus-labs/vanus/proto/pkg/codec"
29+
"google.golang.org/grpc/credentials/insecure"
2430

2531
"go.opentelemetry.io/otel/attribute"
2632
tracesdk "go.opentelemetry.io/otel/sdk/trace"
2733
)
2834

35+
func GetExporter(endpoints []string, eventbus string) tracesdk.SpanExporter {
36+
spanExporter, err := New(context.Background(), WithEndpoints(endpoints), WithEventbus(eventbus))
37+
if err != nil {
38+
log.Error().Err(err).Msg("new span exporter failed")
39+
os.Exit(-1)
40+
}
41+
return spanExporter
42+
}
43+
2944
type Option func(*Options)
3045

3146
type Options struct {
32-
Endpoints string
47+
Endpoints []string
3348
Eventbus string
3449
}
3550

3651
func defaultOptions() *Options {
3752
return &Options{
38-
Endpoints: "127.0.0.1:8080",
53+
Endpoints: []string{},
3954
Eventbus: "event-tracing",
4055
}
4156
}
4257

43-
func WithEndpoint(endpoint string) Option {
58+
func WithEndpoints(endpoints []string) Option {
4459
return func(options *Options) {
45-
options.Endpoints = endpoint
60+
options.Endpoints = endpoints
4661
}
4762
}
4863

@@ -54,9 +69,8 @@ func WithEventbus(eventbus string) Option {
5469

5570
// Exporter exports trace data in the OTLP wire format.
5671
type Exporter struct {
57-
endpoints string
58-
client vanussdk.Client
59-
publisher vanussdk.Publisher
72+
endpoints []string
73+
writer api.BusWriter
6074
}
6175

6276
var _ tracesdk.SpanExporter = (*Exporter)(nil)
@@ -67,47 +81,52 @@ func New(ctx context.Context, opts ...Option) (*Exporter, error) {
6781
apply(defaultOpts)
6882
}
6983

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

80-
ebOpt := vanussdk.WithEventbus("default", defaultOpts.Eventbus)
95+
c := client.Connect(defaultOpts.Endpoints)
96+
bus := c.Eventbus(ctx, api.WithName(defaultOpts.Eventbus), api.WithID(eventbus.Id))
8197
exporter := &Exporter{
8298
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())
99+
writer: bus.Writer(),
89100
}
90101
return exporter, nil
91102
}
92103

93104
// ExportSpans exports a batch of spans.
94105
func (e *Exporter) ExportSpans(ctx context.Context, ss []tracesdk.ReadOnlySpan) error {
95-
es := make([]*v2.Event, 0)
106+
ces := make([]*cloudevents.CloudEvent, 0)
96107
for _, span := range ss {
97-
if span.Name() != "EventTracing" {
108+
event := newEvent(span)
109+
if event.Type() != "event-tracing" {
98110
continue
99111
}
100-
event := newEvent(span)
101-
es = append(es, &event)
112+
eventpb, err := codec.ToProto(&event)
113+
if err != nil {
114+
log.Error(ctx).Err(err).Any("event", event).Msg("failed to proto event")
115+
return nil
116+
}
117+
ces = append(ces, eventpb)
102118
}
103119

104-
if len(es) == 0 {
120+
if len(ces) == 0 {
105121
return nil
106122
}
107123

108-
err := e.publisher.Publish(ctx, es...)
124+
ceBatch := &cloudevents.CloudEventBatch{
125+
Events: ces,
126+
}
127+
_, err := e.writer.Append(ctx, ceBatch)
109128
if err != nil {
110-
log.Error(ctx).Err(err).Msg("failed to publish events to tracing eventbus")
129+
log.Error(ctx).Err(err).Msg("failed to append events to tracing eventbus")
111130
return nil
112131
}
113132
return nil
@@ -122,7 +141,6 @@ func newEvent(span tracesdk.ReadOnlySpan) v2.Event {
122141
event := v2.NewEvent()
123142
event.SetID(uuid.New().String())
124143
event.SetSource(span.Name())
125-
event.SetType(span.SpanKind().String())
126144
data := make(map[string]interface{})
127145
data["name"] = span.Name()
128146
data["trace_id"] = span.SpanContext().TraceID().String()
@@ -136,6 +154,9 @@ func newEvent(span tracesdk.ReadOnlySpan) v2.Event {
136154
data[string(attr.Key)] = attr.Value.AsInt64()
137155
} else if attr.Value.Type() == attribute.STRING {
138156
data[string(attr.Key)] = attr.Value.AsString()
157+
if string(attr.Key) == "type" && attr.Value.AsString() == "event-tracing" {
158+
event.SetType("event-tracing")
159+
}
139160
}
140161
}
141162
data["events"] = span.Events()

‎cmd/gateway/main.go‎

Lines changed: 2 additions & 1 deletion
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‎

Lines changed: 2 additions & 1 deletion
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‎

Lines changed: 1 addition & 3 deletions
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‎

Lines changed: 0 additions & 2 deletions
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‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -106,8 +106,9 @@ 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+
event.SetExtension("traceparent", fmt.Sprintf("00-%s-%s-%s", span.SpanContext().TraceID().String(), span.SpanContext().SpanID().String(), span.SpanContext().TraceFlags().String()))
110+
span.SetName(event.ID())
111+
span.SetAttributes(attribute.String("type", "event-tracing"))
111112
span.SetAttributes(attribute.String("event_id", event.ID()))
112113
span.AddEvent("received from source", trace.WithTimestamp(time.Now()))
113114

‎internal/trigger/reader/reader.go‎

Lines changed: 7 additions & 4 deletions
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‎

Lines changed: 1 addition & 16 deletions
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)