feat: open tracing - alpha

cli/serve.go

@@ -17,11 +17,14 @@ package cli
 
 import (
 	"context"
+	"fmt"
 	"os"
 
 	"github.com/spf13/cobra"
 	"github.com/yomorun/yomo"
+	pkgconfig "github.com/yomorun/yomo/pkg/config"
 	"github.com/yomorun/yomo/pkg/log"
+	"github.com/yomorun/yomo/pkg/trace"
 )
 
 // serveCmd represents the serve command
@@ -36,8 +39,37 @@ var serveCmd = &cobra.Command{
 		}
 
 		log.InfoStatusEvent(os.Stdout, "Running YoMo-Zipper...")
+		// config
+		conf, err := pkgconfig.ParseConfigFile(config)
+		if err != nil {
+			log.FailureStatusEvent(os.Stdout, err.Error())
+			return
+		}
+		ctx := context.Background()
+		// trace
+		tp, shutdown, err := trace.NewTracerProviderWithJaeger("yomo-zipper")
+		if err == nil {
+			log.InfoStatusEvent(os.Stdout, "[zipper] 🛰 trace enabled")
+		}
+		defer shutdown(ctx)
+		// listening address.
+		listenAddr := fmt.Sprintf("%s:%d", conf.Host, conf.Port)
+
+		options := []yomo.ZipperOption{yomo.WithZipperTracerProvider(tp)}
+		if _, ok := conf.Auth["type"]; ok {
+			if tokenString, ok := conf.Auth["token"]; ok {
+				options = append(options, yomo.WithAuth("token", tokenString))
+			}
+		}
+
+		zipper, err := yomo.NewZipper(conf.Name, conf.Functions, conf.Downstreams, options...)
+		if err != nil {
+			log.FailureStatusEvent(os.Stdout, err.Error())
+			return
+		}
+		zipper.Logger().Info("using config file", "file_path", config)
 
-		err := yomo.RunZipper(context.Background(), config)
+		err = zipper.ListenAndServe(ctx, listenAddr)
 		if err != nil {
 			log.FailureStatusEvent(os.Stdout, err.Error())
 			return

cli/serverless/deno/runtime.go

@@ -2,6 +2,7 @@
 package deno
 
 import (
+	"context"
 	"encoding/binary"
 	"errors"
 	"io"
@@ -14,6 +15,7 @@ import (
 	"github.com/yomorun/yomo"
 	"github.com/yomorun/yomo/core/frame"
 	"github.com/yomorun/yomo/pkg/file"
+	"github.com/yomorun/yomo/pkg/trace"
 	"github.com/yomorun/yomo/serverless"
 )
 
@@ -88,10 +90,17 @@ func runDeno(jsPath string, socketPath string, errCh chan<- error) {
 }
 
 func startSfn(name string, zipperAddr string, credential string, observed []frame.Tag, conn net.Conn, errCh chan<- error) (yomo.StreamFunction, error) {
+	// trace
+	tp, shutdown, err := trace.NewTracerProviderWithJaeger("yomo-sfn")
+	if err == nil {
+		log.Println("[sfn] 🛰 trace enabled")
+	}
+	defer shutdown(context.Background())
 	sfn := yomo.NewStreamFunction(
 		name,
 		zipperAddr,
 		yomo.WithSfnCredential(credential),
+		yomo.WithSfnTracerProvider(tp),
 	)
 
 	sfn.SetObserveDataTags(observed...)
@@ -154,7 +163,7 @@ func startSfn(name string, zipperAddr string, credential string, observed []fram
 		},
 	)
 
-	err := sfn.Connect()
+	err = sfn.Connect()
 	if err != nil {
 		return nil, err
 	}

cli/serverless/golang/templates/main.tmpl

@@ -24,17 +24,22 @@ func main() {
 }
 
 func runSFN(name string, addr string, credential string) (closeFn func() error, err error) {
+	// trace
+	tp, shutdown, err := trace.NewTracerProviderWithJaeger("yomo-sfn")
+	if err == nil {
+		log.Println("[sfn] 🛰 trace enabled")
+	}
+	defer shutdown(context.Background())
 	sfn := yomo.NewStreamFunction(
-        name,
+    name,
 		addr,
 		yomo.WithCredential(credential),
+		yomo.WithTracerProvider(tp),
 	)
 	closeFn = sfn.Close
-  // set observe data tags
-  sfn.SetObserveDataTags(DataTags()...)
 
-  	// set observe data tags
-  	sfn.SetObserveDataTags(DataTags()...)
+	// set observe data tags
+	sfn.SetObserveDataTags(DataTags()...)
 
 	// set handler
 	sfn.SetHandler(Handler)

cli/serverless/golang/templates/main_rx.tmpl

@@ -24,14 +24,19 @@ func main() {
 }
 
 func runSFN(name string, addr string, credential string) (closeFn func() error, err error) {
+	// trace
+	tp, shutdown, err := trace.NewTracerProviderWithJaeger("yomo-sfn")
+	if err == nil {
+		log.Println("[sfn] 🛰 trace enabled")
+	}
+	defer shutdown(context.Background())
 	sfn := yomo.NewStreamFunction(
 		name,
 		addr,
 		yomo.WithCredential(credential),
+		yomo.WithTracerProvider(tp),
 	)
 	closeFn = sfn.Close
-  // set observe data tags
-  sfn.SetObserveDataTags(DataTags()...)
 
 	// set observe data tags
 	sfn.SetObserveDataTags(DataTags()...)

cli/serverless/wasm/serverless.go

@@ -2,13 +2,15 @@
 package wasm
 
 import (
+	"context"
 	"log"
 	"os"
 	"sync"
 
 	"github.com/yomorun/yomo"
 	cli "github.com/yomorun/yomo/cli/serverless"
 	pkglog "github.com/yomorun/yomo/pkg/log"
+	"github.com/yomorun/yomo/pkg/trace"
 	"github.com/yomorun/yomo/serverless"
 )
 
@@ -52,12 +54,19 @@ func (s *wasmServerless) Build(clean bool) error {
 // Run the wasm serverless function
 func (s *wasmServerless) Run(verbose bool) error {
 	var wg sync.WaitGroup
+	// trace
+	tp, shutdown, err := trace.NewTracerProviderWithJaeger("yomo-sfn")
+	if err == nil {
+		pkglog.InfoStatusEvent(os.Stdout, "[sfn] 🛰 trace enabled")
+	}
+	defer shutdown(context.Background())
 
 	for _, addr := range s.zipperAddrs {
 		sfn := yomo.NewStreamFunction(
 			s.name,
 			addr,
 			yomo.WithSfnCredential(s.credential),
+			yomo.WithSfnTracerProvider(tp),
 		)
 		sfn.SetObserveDataTags(s.observed...)
 

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", "sid").Encode()
+	mdBytes, _ := NewDefaultMetadata(source.clientID, true, "tid", "sid", false).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", "sid"),
+		NewDefaultMetadata(source.clientID, true, "tid", "sid", false),
 		metadata.M{
 			"foo": "bar",
 		},

core/metadata.go

@@ -7,19 +7,25 @@ const (
 	MetadataBroadcastKey = "yomo-broadcast"
 	MetadataTIDKey       = "yomo-tid"
 	MetadataSIDKey       = "yomo-sid"
+	MetaTraced           = "yomo-traced"
 )
 
 // NewDefaultMetadata returns a default metadata.
-func NewDefaultMetadata(sourceID string, broadcast bool, tid string, sid string) metadata.M {
+func NewDefaultMetadata(sourceID string, broadcast bool, tid string, sid string, traced bool) metadata.M {
 	broadcastString := "false"
 	if broadcast {
 		broadcastString = "true"
 	}
+	tracedString := "false"
+	if traced {
+		tracedString = "true"
+	}
 	return metadata.M{
 		MetadataSourceIDKey:  sourceID,
 		MetadataBroadcastKey: broadcastString,
 		MetadataTIDKey:       tid,
 		MetadataSIDKey:       sid,
+		MetaTraced:           tracedString,
 	}
 }
 
@@ -47,6 +53,12 @@ func GetSIDFromMetadata(m metadata.M) string {
 	return sid
 }
 
+// GetTracedFromMetadata gets traced from metadata.
+func GetTracedFromMetadata(m metadata.M) bool {
+	traced, _ := m.Get(MetaTraced)
+	return traced == "true"
+}
+
 // SetTIDToMetadata sets tid to metadata.
 func SetTIDToMetadata(m metadata.M, tid string) {
 	m.Set(MetadataTIDKey, tid)
@@ -56,3 +68,12 @@ func SetTIDToMetadata(m metadata.M, tid string) {
 func SetSIDToMetadata(m metadata.M, sid string) {
 	m.Set(MetadataSIDKey, sid)
 }
+
+// SetTracedToMetadata sets traced to metadata.
+func SetTracedToMetadata(m metadata.M, traced bool) {
+	tracedString := "false"
+	if traced {
+		tracedString = "true"
+	}
+	m.Set(MetaTraced, tracedString)
+}

core/metadata_test.go

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

core/server.go

@@ -319,18 +319,29 @@ func (s *Server) handleDataFrame(c *Context) error {
 	atomic.AddInt64(&s.counterOfDataFrame, 1)
 
 	from := c.DataStream
-	// trace
 	tid := GetTIDFromMetadata(c.FrameMetadata)
 	sid := GetSIDFromMetadata(c.FrameMetadata)
+	parentTraced := GetTracedFromMetadata(c.FrameMetadata)
+	traced := false
+	// trace
 	tp := s.TracerProvider()
 	if tp != nil {
-		span, err := trace.NewSpan(tp, "zipper", "handle DataFrame", tid, sid)
+		// create span
+		var span oteltrace.Span
+		var err error
+		// set parent span, if not traced, use empty string
+		if parentTraced {
+			span, err = trace.NewSpan(tp, "zipper", "handle DataFrame", tid, sid)
+		} else {
+			span, err = trace.NewSpan(tp, "zipper", "handle DataFrame", "", "")
+		}
 		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()
+			traced = true
 		}
 	}
 	if tid == "" {
@@ -344,15 +355,14 @@ func (s *Server) handleDataFrame(c *Context) error {
 	// reallocate metadata with new TID and SID
 	SetTIDToMetadata(c.FrameMetadata, tid)
 	SetSIDToMetadata(c.FrameMetadata, sid)
+	SetTracedToMetadata(c.FrameMetadata, traced || parentTraced)
 	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())
-	}
+	s.logger.Debug("zipper metadata", "tid", tid, "sid", sid, "parentTraced", parentTraced, "traced", traced, "frome_stream_name", from.Name())
 	// route
 	route := s.router.Route(c.FrameMetadata)
 	if route == nil {

example/9-cli/Taskfile.yaml

@@ -41,13 +41,13 @@ tasks:
       - "yomo build -m go.mod app.go"
     status:
       - test -f sfn/sfn.wasm
-    internal: true
 
   source:
     desc: run source
     deps: [source-build]
     env:
       YOMO_LOG_LEVEL: error
+      YOMO_TRACE_JAEGER_ENDPOINT: 'http://localhost:14268/api/traces'
     cmds:
       - "./bin/source{{exeExt}}"
 
@@ -67,3 +67,4 @@ tasks:
       - "yomo serve -c config.yaml"
     env:
       YOMO_LOG_LEVEL: error
+      YOMO_TRACE_JAEGER_ENDPOINT: 'http://localhost:14268/api/traces'

example/9-cli/sfn/.env

@@ -1 +1,2 @@
 YOMO_SFN_NAME=Noise
+YOMO_TRACE_JAEGER_ENDPOINT=http://localhost:14268/api/traces

example/9-cli/source/go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/matoous/go-nanoid/v2 v2.0.0 // indirect
 	github.com/onsi/ginkgo/v2 v2.11.0 // indirect
 	github.com/quic-go/qtls-go1-20 v0.3.1 // indirect
-	github.com/quic-go/quic-go v0.37.3 // indirect
+	github.com/quic-go/quic-go v0.37.4 // indirect
 	github.com/vmihailenco/msgpack/v5 v5.3.5 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	github.com/yomorun/y3 v1.0.5 // indirect

example/9-cli/source/go.sum

@@ -32,8 +32,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/quic-go/qtls-go1-20 v0.3.1 h1:O4BLOM3hwfVF3AcktIylQXyl7Yi2iBNVy5QsV+ySxbg=
 github.com/quic-go/qtls-go1-20 v0.3.1/go.mod h1:X9Nh97ZL80Z+bX/gUXMbipO6OxdiDi58b/fMC9mAL+k=
-github.com/quic-go/quic-go v0.37.3 h1:pkHH3xaMNUNAh6OtgEV/0K6Fz+YIJXhPzgd/ShiRDm4=
-github.com/quic-go/quic-go v0.37.3/go.mod h1:YsbH1r4mSHPJcLF4k4zruUkLBqctEMBDR6VPvcYjIsU=
+github.com/quic-go/quic-go v0.37.4 h1:ke8B73yMCWGq9MfrCCAw0Uzdm7GaViC3i39dsIdDlH4=
+github.com/quic-go/quic-go v0.37.4/go.mod h1:YsbH1r4mSHPJcLF4k4zruUkLBqctEMBDR6VPvcYjIsU=
 github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
 github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=