feat: trace (#575)

# Description

Added the OpenTelemetry interface

## Usage

Use Jaeger as the tracer exporter，configure the value of environment a
for your [Jaeger](https://www.jaegertracing.io/) endpoint:

```
YOMO_TRACE_JAEGER_ENDPOINT=http://localhost:14268/api/traces
```


### Source Tracing

```go
// trace
tp, shutdown, err := trace.NewTracerProviderWithJaeger("yomo-source")
if err == nil {
	log.Println("[source] 🛰 trace enabled")
}
defer shutdown(context.Background())
// source
source := yomo.NewSource(
	"source",
	"localhost:9000",
	yomo.WithTracerProvider(tp),  // use tracer provider
)
...
```

### SFN Tracing

```go
// trace
tp, shutdown, err := trace.NewTracerProviderWithJaeger("yomo-sfn")
if err == nil {
	log.Println("[sfn] 🛰 trace enabled")
}
defer shutdown(context.Background())
// source
sfn := yomo.NewStreamFunction(
	"sfn",
	"localhost:9000",
	yomo.WithSfnTracerProvider(tp),  // use tracer provider
)
...
```

### Zipper Tracing

```go
// trace
tp, shutdown, err := trace.NewTracerProviderWithJaeger("yomo-zipper")
if err == nil {
	log.Println("[zipper] 🛰 trace enabled")
}
defer shutdown(context.Background())
// zipper
conf, err := config.ParseConfigFile("config.yaml")
if err != nil {
	log.Fatal(err)
}
zipper := yomo.NewZipper(
	conf.Name,
	conf.Functions,
        conf.Downstreams,
	yomo.WithZipperTracerProvider(tp),  // use tracer provider
)
...
```

---------

Co-authored-by: woorui <rui1009479218@gmail.com>

Author: venjiang

core/client.go

@@ -6,26 +6,29 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"reflect"
 	"runtime"
 	"time"
 
 	"github.com/yomorun/yomo/core/frame"
 	"github.com/yomorun/yomo/pkg/frame-codec/y3codec"
 	"github.com/yomorun/yomo/pkg/id"
+	oteltrace "go.opentelemetry.io/otel/trace"
 	"golang.org/x/exp/slog"
 )
 
 // Client is the abstraction of a YoMo-Client. a YoMo-Client can be
 // Source, Upstream Zipper or StreamFunction.
 type Client struct {
-	name       string                     // name of the client
-	clientID   string                     // id of the client
-	streamType StreamType                 // type of the dataStream
-	processor  func(*frame.DataFrame)     // function to invoke when data arrived
-	receiver   func(*frame.BackflowFrame) // function to invoke when data is processed
-	errorfn    func(error)                // function to invoke when error occured
-	opts       *clientOptions
-	logger     *slog.Logger
+	name           string                     // name of the client
+	clientID       string                     // id of the client
+	streamType     StreamType                 // type of the dataStream
+	processor      func(*frame.DataFrame)     // function to invoke when data arrived
+	receiver       func(*frame.BackflowFrame) // function to invoke when data is processed
+	errorfn        func(error)                // function to invoke when error occured
+	opts           *clientOptions
+	logger         *slog.Logger
+	tracerProvider oteltrace.TracerProvider
 
 	// ctx and ctxCancel manage the lifecycle of client.
 	ctx       context.Context
@@ -57,6 +60,7 @@ func NewClient(appName string, connType ClientType, opts ...ClientOption) *Clien
 		streamType:     connType,
 		opts:           option,
 		logger:         logger,
+		tracerProvider: option.tracerProvider,
 		errorfn:        func(err error) { logger.Error("client err", "err", err) },
 		writeFrameChan: make(chan frame.Frame),
 		ctx:            ctx,
@@ -358,3 +362,14 @@ type FrameWriterConnection interface {
 	Close() error
 	Connect(context.Context, string) error
 }
+
+// TracerProvider returns the tracer provider of client.
+func (c *Client) TracerProvider() oteltrace.TracerProvider {
+	if c.tracerProvider == nil {
+		return nil
+	}
+	if reflect.ValueOf(c.tracerProvider).IsNil() {
+		return nil
+	}
+	return c.tracerProvider
+}

core/client_options.go

@@ -9,6 +9,7 @@ import (
 	"github.com/yomorun/yomo/core/frame"
 	"github.com/yomorun/yomo/core/ylog"
 	pkgtls "github.com/yomorun/yomo/pkg/tls"
+	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/exp/slog"
 )
 
@@ -24,6 +25,7 @@ type clientOptions struct {
 	connectUntilSucceed bool
 	nonBlockWrite       bool
 	logger              *slog.Logger
+	tracerProvider      trace.TracerProvider
 }
 
 func defaultClientOption() *clientOptions {
@@ -103,6 +105,12 @@ func WithLogger(logger *slog.Logger) ClientOption {
 	}
 }
 
+func WithTracerProvider(tp trace.TracerProvider) ClientOption {
+	return func(o *clientOptions) {
+		o.tracerProvider = tp
+	}
+}
+
 // ClientType is equal to StreamType.
 type ClientType = StreamType
 

core/client_test.go

@@ -131,7 +131,7 @@ func TestFrameRoundTrip(t *testing.T) {
 	exited = checkClientExited(sameNameSfn, time.Second)
 	assert.False(t, exited, "the new sfn stream should not exited")
 
-	mdBytes, _ := NewDefaultMetadata(source.clientID, true, "tid").Encode()
+	mdBytes, _ := NewDefaultMetadata(source.clientID, true, "tid", "sid").Encode()
 
 	err = sameNameSfn.WriteFrame(&frame.DataFrame{Tag: backflowTag, Metadata: mdBytes, Payload: backflow})
 	assert.NoError(t, err)
@@ -144,7 +144,7 @@ func TestFrameRoundTrip(t *testing.T) {
 	assert.ElementsMatch(t, nameList, []string{"source", "sfn-1"})
 
 	md := metadata.New(
-		NewDefaultMetadata(source.clientID, true, "tid"),
+		NewDefaultMetadata(source.clientID, true, "tid", "sid"),
 		metadata.M{
 			"foo": "bar",
 		},

core/metadata.go

@@ -2,39 +2,57 @@ package core
 
 import "github.com/yomorun/yomo/core/metadata"
 
-var (
-	metadataSourceIDKey  = "yomo-source-id"
-	metadataBroadcastKey = "yomo-broadcast"
-	metadataTIDKey       = "yomo-tid"
+const (
+	MetadataSourceIDKey  = "yomo-source-id"
+	MetadataBroadcastKey = "yomo-broadcast"
+	MetadataTIDKey       = "yomo-tid"
+	MetadataSIDKey       = "yomo-sid"
 )
 
 // NewDefaultMetadata returns a default metadata.
-func NewDefaultMetadata(sourceID string, broadcast bool, tid string) metadata.M {
+func NewDefaultMetadata(sourceID string, broadcast bool, tid string, sid string) metadata.M {
 	broadcastString := "false"
 	if broadcast {
 		broadcastString = "true"
 	}
 	return metadata.M{
-		metadataSourceIDKey:  sourceID,
-		metadataBroadcastKey: broadcastString,
-		metadataTIDKey:       tid,
+		MetadataSourceIDKey:  sourceID,
+		MetadataBroadcastKey: broadcastString,
+		MetadataTIDKey:       tid,
+		MetadataSIDKey:       sid,
 	}
 }
 
 // GetSourceIDFromMetadata gets sourceID from metadata.
 func GetSourceIDFromMetadata(m metadata.M) string {
-	sourceID, _ := m.Get(metadataSourceIDKey)
+	sourceID, _ := m.Get(MetadataSourceIDKey)
 	return sourceID
 }
 
 // GetBroadcastFromMetadata gets broadcast from metadata.
 func GetBroadcastFromMetadata(m metadata.M) bool {
-	broadcast, _ := m.Get(metadataBroadcastKey)
+	broadcast, _ := m.Get(MetadataBroadcastKey)
 	return broadcast == "true"
 }
 
 // GetTIDFromMetadata gets TID from metadata.
 func GetTIDFromMetadata(m metadata.M) string {
-	tid, _ := m.Get(metadataTIDKey)
+	tid, _ := m.Get(MetadataTIDKey)
 	return tid
 }
+
+// GetSIDFromMetadata gets SID from metadata.
+func GetSIDFromMetadata(m metadata.M) string {
+	sid, _ := m.Get(MetadataSIDKey)
+	return sid
+}
+
+// SetTIDToMetadata sets tid to metadata.
+func SetTIDToMetadata(m metadata.M, tid string) {
+	m.Set(MetadataTIDKey, tid)
+}
+
+// SetSIDToMetadata sets sid to metadata.
+func SetSIDToMetadata(m metadata.M, sid string) {
+	m.Set(MetadataSIDKey, sid)
+}

core/metadata_test.go

@@ -7,9 +7,16 @@ import (
 )
 
 func TestMetadata(t *testing.T) {
-	md := NewDefaultMetadata("source", true, "xxxxxxx")
+	md := NewDefaultMetadata("source", true, "xxxxxxx", "sssssss")
 
 	assert.Equal(t, "source", GetSourceIDFromMetadata(md))
 	assert.Equal(t, true, GetBroadcastFromMetadata(md))
 	assert.Equal(t, "xxxxxxx", GetTIDFromMetadata(md))
+	assert.Equal(t, "sssssss", GetSIDFromMetadata(md))
+
+	SetTIDToMetadata(md, "ccccccc")
+	assert.Equal(t, "ccccccc", GetTIDFromMetadata(md))
+
+	SetSIDToMetadata(md, "aaaaaaa")
+	assert.Equal(t, "aaaaaaa", GetSIDFromMetadata(md))
 }

core/server.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"reflect"
 	"sync"
 	"sync/atomic"
 
@@ -21,6 +22,9 @@ import (
 	// authentication implements, Currently, only token authentication is implemented
 	_ "github.com/yomorun/yomo/pkg/auth"
 	"github.com/yomorun/yomo/pkg/frame-codec/y3codec"
+	"github.com/yomorun/yomo/pkg/id"
+	"github.com/yomorun/yomo/pkg/trace"
+	oteltrace "go.opentelemetry.io/otel/trace"
 )
 
 // ErrServerClosed is returned by the Server's Serve and ListenAndServe methods after a call to Shutdown or Close.
@@ -51,6 +55,7 @@ type Server struct {
 	connectionCloseHandlers []ConnectionHandler
 	listener                Listener
 	logger                  *slog.Logger
+	tracerProvider          oteltrace.TracerProvider
 }
 
 // NewServer create a Server instance.
@@ -71,6 +76,7 @@ func NewServer(name string, opts ...ServerOption) *Server {
 		name:             name,
 		downstreams:      make(map[string]FrameWriterConnection),
 		logger:           logger,
+		tracerProvider:   options.tracerProvider,
 		codec:            y3codec.Codec(),
 		packetReadWriter: y3codec.PacketReadWriter(),
 		opts:             options,
@@ -313,7 +319,40 @@ func (s *Server) handleDataFrame(c *Context) error {
 	atomic.AddInt64(&s.counterOfDataFrame, 1)
 
 	from := c.DataStream
-
+	// trace
+	tid := GetTIDFromMetadata(c.FrameMetadata)
+	sid := GetSIDFromMetadata(c.FrameMetadata)
+	tp := s.TracerProvider()
+	if tp != nil {
+		span, err := trace.NewSpan(tp, "zipper", "handle DataFrame", tid, sid)
+		if err != nil {
+			s.logger.Error("zipper trace error", "err", err)
+		} else {
+			defer span.End()
+			tid = span.SpanContext().TraceID().String()
+			sid = span.SpanContext().SpanID().String()
+		}
+	}
+	if tid == "" {
+		s.logger.Debug("zipper create new tid")
+		tid = id.TID()
+	}
+	if sid == "" {
+		s.logger.Debug("zipper create new sid")
+		sid = id.SID()
+	}
+	// reallocate metadata with new TID and SID
+	SetTIDToMetadata(c.FrameMetadata, tid)
+	SetSIDToMetadata(c.FrameMetadata, sid)
+	md, err := c.FrameMetadata.Encode()
+	if err != nil {
+		s.logger.Error("encode metadata error", "err", err)
+		return err
+	}
+	c.Frame.Metadata = md
+	if tp != nil {
+		s.logger.Debug("zipper trace", "tid", tid, "sid", sid, "frome_stream_name", from.Name())
+	}
 	// route
 	route := s.router.Route(c.FrameMetadata)
 	if route == nil {
@@ -435,6 +474,7 @@ func (s *Server) dispatchToDownstreams(c *Context) {
 		var (
 			broadcast = GetBroadcastFromMetadata(c.FrameMetadata)
 			tid       = GetTIDFromMetadata(c.FrameMetadata)
+			sid       = GetSIDFromMetadata(c.FrameMetadata)
 		)
 		if broadcast {
 			mdBytes, err := c.FrameMetadata.Encode()
@@ -445,7 +485,7 @@ func (s *Server) dispatchToDownstreams(c *Context) {
 			c.Frame.Metadata = mdBytes
 
 			for streamID, ds := range s.downstreams {
-				c.Logger.Info("dispatching to downstream", "dispatch_stream_id", streamID, "tid", tid)
+				c.Logger.Info("dispatching to downstream", "dispatch_stream_id", streamID, "tid", tid, "sid", sid)
 				ds.WriteFrame(c.Frame)
 			}
 		}
@@ -493,3 +533,14 @@ func (s *Server) authNames() []string {
 
 // Name returns the name of server.
 func (s *Server) Name() string { return s.name }
+
+// TracerProvider returns the tracer provider of server.
+func (s *Server) TracerProvider() oteltrace.TracerProvider {
+	if s.tracerProvider == nil {
+		return nil
+	}
+	if reflect.ValueOf(s.tracerProvider).IsNil() {
+		return nil
+	}
+	return s.tracerProvider
+}

core/server_options.go

@@ -6,6 +6,7 @@ import (
 	"github.com/quic-go/quic-go"
 	"github.com/yomorun/yomo/core/auth"
 	"github.com/yomorun/yomo/core/ylog"
+	oteltrace "go.opentelemetry.io/otel/trace"
 	"golang.org/x/exp/slog"
 )
 
@@ -15,10 +16,11 @@ type ServerOption func(*serverOptions)
 // ServerOptions are the options for YoMo server.
 // TODO: quic alpn function.
 type serverOptions struct {
-	quicConfig *quic.Config
-	tlsConfig  *tls.Config
-	auths      map[string]auth.Authentication
-	logger     *slog.Logger
+	quicConfig     *quic.Config
+	tlsConfig      *tls.Config
+	auths          map[string]auth.Authentication
+	logger         *slog.Logger
+	tracerProvider oteltrace.TracerProvider
 }
 
 func defaultServerOptions() *serverOptions {
@@ -66,3 +68,10 @@ func WithServerLogger(logger *slog.Logger) ServerOption {
 		o.logger = logger
 	}
 }
+
+// WithServerTracerProvider sets tracer provider for the server.
+func WithServerTracerProvider(tp oteltrace.TracerProvider) ServerOption {
+	return func(o *serverOptions) {
+		o.tracerProvider = tp
+	}
+}

core/serverless/context.go

@@ -1,6 +1,8 @@
 package serverless
 
-import "github.com/yomorun/yomo/core/frame"
+import (
+	"github.com/yomorun/yomo/core/frame"
+)
 
 // Context sfn handler context
 type Context struct {

example/0-basic/Taskfile.yml

@@ -22,16 +22,19 @@ tasks:
     deps: [source-build, sfn-build]
     cmds:
       - echo 'building done'
+    internal: true
 
   source-build:
     desc: build source
     cmds:
       - "go build -o ./bin/source{{exeExt}} source/main.go"
+    internal: true
 
   sfn-build:
     desc: build sfn
     cmds:
       - "go build -o ./bin/sfn{{exeExt}} sfn/main.go"
+    internal: true
 
   source:
     desc: run source

example/0-basic/sfn/main.go

@@ -32,12 +32,12 @@ func main() {
 	// start
 	err := sfn.Connect()
 	if err != nil {
-		slog.Error("[sfn1] connect", err)
+		slog.Error("[sfn] connect", err)
 		os.Exit(1)
 	}
 	// set the error handler function when server error occurs
 	sfn.SetErrorHandler(func(err error) {
-		slog.Error("[sfn1] receive server error", "err", err)
+		slog.Error("[sfn] receive server error", "err", err)
 		sfn.Close()
 		os.Exit(1)
 	})