feat: support export trace to collector

etc/kuiper.yaml

@@ -139,4 +139,8 @@ portable:
   # or other circumstance where the python executable cannot be successfully invoked through the default command.
   pythonBin: python
   # control init timeout. If the init time is longer than this value, the plugin will be terminated.
-  initTimeout: 5s
+  initTimeout: 5s
+
+openTelemetry:
+  enableCollector: false
+  endpoint: localhost:4318

go.mod

@@ -57,14 +57,15 @@ require (
 	github.com/yisaer/file-rotatelogs v0.0.0-20240606072000-fe792f3c4e04
 	go.nanomsg.org/mangos/v3 v3.4.2
 	go.opentelemetry.io/otel v1.28.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.28.0
 	go.opentelemetry.io/otel/sdk v1.28.0
 	go.opentelemetry.io/otel/trace v1.28.0
 	go.uber.org/automaxprocs v1.5.3
 	golang.org/x/exp v0.0.0-20240318143956-a85f2c67cd81
-	golang.org/x/text v0.14.0
-	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237
-	google.golang.org/grpc v1.62.1
-	google.golang.org/protobuf v1.33.0
+	golang.org/x/text v0.16.0
+	google.golang.org/genproto/googleapis/api v0.0.0-20240701130421-f6361c86f094
+	google.golang.org/grpc v1.64.0
+	google.golang.org/protobuf v1.34.2
 	gopkg.in/ini.v1 v1.67.0
 	gopkg.in/yaml.v3 v3.0.1
 	modernc.org/sqlite v1.29.5
@@ -104,6 +105,7 @@ require (
 	github.com/go-redis/redis/v7 v7.4.1 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
 	github.com/google/pprof v0.0.0-20240528025155-186aa0362fba // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
@@ -138,7 +140,6 @@ require (
 	github.com/prometheus/common v0.50.0 // indirect
 	github.com/prometheus/procfs v0.13.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
-	github.com/rogpeppe/go-internal v1.12.0 // indirect
 	github.com/rs/xid v1.4.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/segmentio/encoding v0.3.6 // indirect
@@ -154,15 +155,17 @@ require (
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/yuin/gopher-lua v1.1.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.28.0 // indirect
 	go.opentelemetry.io/otel/metric v1.28.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.uber.org/goleak v1.3.0 // indirect
-	golang.org/x/crypto v0.21.0 // indirect
-	golang.org/x/mod v0.16.0 // indirect
-	golang.org/x/net v0.23.0 // indirect
-	golang.org/x/sync v0.6.0 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
+	golang.org/x/mod v0.17.0 // indirect
+	golang.org/x/net v0.26.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/sys v0.21.0 // indirect
-	golang.org/x/tools v0.19.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
+	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240701130421-f6361c86f094 // indirect
 	gopkg.in/src-d/go-errors.v1 v1.0.0 // indirect
 	modernc.org/gc/v3 v3.0.0-20240304020402-f0dba7c97c2b // indirect
 	modernc.org/libc v1.45.2 // indirect
@@ -172,4 +175,9 @@ require (
 	modernc.org/token v1.1.0 // indirect
 )
 
+replace (
+	google.golang.org/genproto => google.golang.org/genproto v0.0.0-20240701130421-f6361c86f094
+	google.golang.org/genproto/googleapis/rpc => google.golang.org/genproto/googleapis/rpc v0.0.0-20240701130421-f6361c86f094
+)
+
 go 1.22.1

internal/conf/conf.go

@@ -224,9 +224,16 @@
 	Connection struct {
 		BackoffMaxElapsedDuration cast.DurationConf `yaml:"backoffMaxElapsedDuration"`
 	}
+	OpenTelemetry OpenTelemetry `yaml:"openTelemetry"`
+
 	AesKey []byte
 }
 
+type OpenTelemetry struct {
+	EnableCollector bool   `yaml:"enableCollector"`
+	Endpoint        string `yaml:"endpoint"`
+}
+
 func SetLogLevel(level string, debug bool) {
 	if debug {
 		Log.SetLevel(logrus.DebugLevel)
@@ -414,6 +421,10 @@
 		_ = Config.Basic.Syslog.Validate()
 	}
 
+	if Config.OpenTelemetry.Endpoint == "" {
+		Config.OpenTelemetry.Endpoint = "localhost:4318"
+	}
+
 	_ = ValidateRuleOption(&Config.Rule)
 }
 

internal/server/server.go

@@ -53,6 +53,7 @@
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/connection"
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+	"github.com/lf-edge/ekuiper/v2/pkg/tracer"
 )
 
 var (
@@ -166,6 +167,9 @@
 	if err := bump.BumpToCurrentVersion(dataDir); err != nil {
 		panic(err)
 	}
+	if err := tracer.InitTracer(); err != nil {
+		conf.Log.Warn(err)
+	}
 
 	keyedstate.InitKeyedStateKV()
 

internal/service/executors.go

@@ -132,7 +132,7 @@ func (d *grpcExecutor) InvokeFunction(_ api.FunctionContext, name string, params
 		)
 		go infra.SafeRun(func() error {
 			defer cancel()
-			conn, e = grpc.DialContext(dialCtx, d.addr.Host, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
+			conn, e = grpc.DialContext(dialCtx, d.addr.Host, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock()) //nolint:staticcheck
 			return e
 		})
 

internal/topo/topo.go

@@ -422,6 +422,7 @@ func (s *Topo) WaitClose() {
 	// wait all operators close
 	if s.opsWg != nil {
 		s.opsWg.Wait()
+		s.opsWg = nil
 		conf.Log.Infof("rule %s stopped", s.ctx.GetRuleId())
 	}
 }

internal/topo/topotest/rule_test.go

@@ -20,10 +20,12 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/def"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/planner"
 	"github.com/lf-edge/ekuiper/v2/internal/topo/topotest/mocknode"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/tracer"
 )
 
 func TestSharedSourceSchemaless(t *testing.T) {
@@ -529,6 +531,8 @@ func TestSRFSQL(t *testing.T) {
 }
 
 func TestSingleSQL(t *testing.T) {
+	conf.InitConf()
+	tracer.InitTracer()
 	// Reset
 	streamList := []string{"demo", "demoError", "demo1", "table1", "demoTable", "demoArr"}
 	HandleStream(false, streamList, t)

pkg/tracer/tracer.go

@@ -15,21 +15,38 @@
 package tracer
 
 import (
+	"context"
+
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
 	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
 	"go.opentelemetry.io/otel/trace"
+
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
 )
 
-func init() {
-	tp := sdktrace.NewTracerProvider(
-		sdktrace.WithResource(resource.NewWithAttributes(
-			semconv.SchemaURL,
-			semconv.ServiceNameKey.String("kuiperd-service"),
-		)),
-	)
+func InitTracer() error {
+	var opts []sdktrace.TracerProviderOption
+	opts = append(opts, sdktrace.WithResource(resource.NewWithAttributes(
+		semconv.SchemaURL,
+		semconv.ServiceNameKey.String("kuiperd-service"),
+	)))
+	if conf.Config.OpenTelemetry.EnableCollector {
+		ctx := context.Background()
+		exporter, err := otlptracehttp.New(ctx,
+			otlptracehttp.WithEndpoint(conf.Config.OpenTelemetry.Endpoint),
+			otlptracehttp.WithInsecure(),
+		)
+		if err != nil {
+			return err
+		}
+		opts = append(opts, sdktrace.WithBatcher(exporter))
+	}
+	tp := sdktrace.NewTracerProvider(opts...)
 	otel.SetTracerProvider(tp)
+	return nil
 }
 
 func GetTracer() trace.Tracer {