From 1778c5ac7bae0297d06af52963b17c6eb92a226d Mon Sep 17 00:00:00 2001 From: wizardchen Date: Mon, 8 Jun 2026 13:19:24 +0800 Subject: [PATCH] refactor(tracing): remove Jaeger integration and related components This commit removes Jaeger tracing support from the project, including the associated Docker configurations, Go modules, and middleware. The changes simplify the codebase by eliminating unused tracing functionality, which was previously integrated for observability. Documentation and scripts have also been updated to reflect the removal of Jaeger references. --- cmd/desktop/main.go | 2 - cmd/server/main.go | 2 - docker-compose.dev.yml | 26 ---- docker-compose.yml | 32 ----- docs/QA.md | 2 +- docs/wiki/开发部署/开发指南.md | 1 - docs/开发指南.md | 3 - go.mod | 11 +- go.sum | 2 - helm/README.md | 1 - helm/templates/NOTES.txt | 5 +- helm/values.yaml | 9 -- .../application/service/knowledge_process.go | 24 ---- .../service/retriever/composite.go | 27 +--- internal/container/container.go | 29 ----- internal/im/service.go | 46 +------ internal/middleware/trace.go | 117 ------------------ internal/router/router.go | 3 - internal/tracing/init.go | 105 ---------------- scripts/cloud-image/README.md | 1 - scripts/cloud-image/prepare.sh | 2 +- scripts/dev.sh | 14 +-- scripts/quick-dev.sh | 1 - scripts/start_all.sh | 2 - 24 files changed, 14 insertions(+), 453 deletions(-) delete mode 100644 internal/middleware/trace.go delete mode 100644 internal/tracing/init.go diff --git a/cmd/desktop/main.go b/cmd/desktop/main.go index 02bb5830..b5c0efd1 100644 --- a/cmd/desktop/main.go +++ b/cmd/desktop/main.go @@ -29,7 +29,6 @@ import ( "github.com/Tencent/WeKnora/internal/container" "github.com/Tencent/WeKnora/internal/logger" "github.com/Tencent/WeKnora/internal/runtime" - "github.com/Tencent/WeKnora/internal/tracing" "github.com/Tencent/WeKnora/internal/types/interfaces" "github.com/joho/godotenv" wailsruntime "github.com/wailsapp/wails/v2/pkg/runtime" @@ -191,7 +190,6 @@ func main() { err := c.Invoke(func( cfg *config.Config, router *gin.Engine, - tracer *tracing.Tracer, resourceCleaner interfaces.ResourceCleaner, ) error { server := &http.Server{Handler: router} diff --git a/cmd/server/main.go b/cmd/server/main.go index fc7efdc1..069fcde1 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -36,7 +36,6 @@ import ( "github.com/Tencent/WeKnora/internal/container" "github.com/Tencent/WeKnora/internal/logger" "github.com/Tencent/WeKnora/internal/runtime" - "github.com/Tencent/WeKnora/internal/tracing" "github.com/Tencent/WeKnora/internal/types/interfaces" ) @@ -66,7 +65,6 @@ func main() { err := c.Invoke(func( cfg *config.Config, router *gin.Engine, - tracer *tracing.Tracer, resourceCleaner interfaces.ResourceCleaner, systemSettingSvc interfaces.SystemSettingService, ) error { diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index b2a2c56f..fa2dbb0d 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -298,31 +298,6 @@ services: - WeKnora-network-dev restart: unless-stopped - jaeger: - image: jaegertracing/all-in-one:latest - container_name: WeKnora-jaeger-dev - ports: - - "6831:6831/udp" - - "6832:6832/udp" - - "5778:5778" - - "16686:16686" - - "4317:4317" - - "4318:4318" - - "14250:14250" - - "14268:14268" - - "9411:9411" - environment: - - COLLECTOR_OTLP_ENABLED=true - - COLLECTOR_ZIPKIN_HOST_PORT=:9411 - volumes: - - jaeger_data_dev:/var/lib/jaeger - networks: - - WeKnora-network-dev - restart: unless-stopped - profiles: - - jaeger - - full - dex: image: dexidp/dex:latest container_name: WeKnora-dex-dev @@ -547,7 +522,6 @@ volumes: redis_data_dev: minio_data_dev: neo4j-data-dev: - jaeger_data_dev: qdrant_data_dev: opensearch_data_dev: milvus_data_dev: diff --git a/docker-compose.yml b/docker-compose.yml index 685cc649..7da358c6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -82,12 +82,6 @@ services: - DB_NAME=${DB_NAME:-} - TZ=${TZ:-Asia/Shanghai} - WEKNORA_LANGUAGE=${WEKNORA_LANGUAGE:-zh-CN} - - OTEL_EXPORTER_OTLP_ENDPOINT=jaeger:4317 - - OTEL_SERVICE_NAME=WeKnora - - OTEL_TRACES_EXPORTER=otlp - - OTEL_METRICS_EXPORTER=none - - OTEL_LOGS_EXPORTER=none - - OTEL_PROPAGATORS=tracecontext,baggage # ========== Langfuse (optional observability) ========== # Enabled automatically when LANGFUSE_PUBLIC_KEY + LANGFUSE_SECRET_KEY are set. # See docs/Langfuse集成.md for the full list of tuning knobs. @@ -398,31 +392,6 @@ services: - minio - full - jaeger: - image: jaegertracing/all-in-one:1.76.0 - container_name: WeKnora-jaeger - ports: - - "6831:6831/udp" # Jaeger Thrift接收器 - - "6832:6832/udp" # Jaeger Thrift接收器(Compact) - - "5778:5778" # 配置端口 - - "16686:16686" # Web UI - - "4317:4317" # OTLP gRPC接收器 - - "4318:4318" # OTLP HTTP接收器 - - "14250:14250" # 接收模型端口 - - "14268:14268" # Jaeger HTTP接收器 - - "9411:9411" # Zipkin兼容性端口 - environment: - - COLLECTOR_OTLP_ENABLED=true - - COLLECTOR_ZIPKIN_HOST_PORT=:9411 - volumes: - - jaeger_data:/var/lib/jaeger # 持久化 Jaeger 数据 - networks: - - WeKnora-network - restart: unless-stopped - profiles: - - jaeger - - full - neo4j: image: neo4j:2025.10.1 container_name: WeKnora-neo4j @@ -822,7 +791,6 @@ volumes: postgres-data: data-files: docreader-tmp: - jaeger_data: minio_data: neo4j-data: qdrant_data: diff --git a/docs/QA.md b/docs/QA.md index 325710ca..b817bb5c 100644 --- a/docs/QA.md +++ b/docs/QA.md @@ -73,7 +73,7 @@ INIT_RERANK_MODEL_API_KEY=your_rerank_model_api_key # 启动 MinIO 服务 docker-compose --profile minio up -d -# 或者启动完整服务（包括 MinIO、Jaeger、Neo4j、Qdrant） +# 或者启动完整服务（包括 MinIO、Neo4j、Qdrant） docker-compose --profile full up -d ``` diff --git a/docs/wiki/开发部署/开发指南.md b/docs/wiki/开发部署/开发指南.md index af0143fb..de178acb 100644 --- a/docs/wiki/开发部署/开发指南.md +++ b/docs/wiki/开发部署/开发指南.md @@ -44,7 +44,6 @@ make dev-frontend | Redis | localhost:6379 | | MinIO Console | http://localhost:9001 | | Neo4j Browser | http://localhost:7474 | -| Jaeger UI | http://localhost:16686 | > Neo4j 的使用参见 [知识图谱](../核心功能/知识图谱.md) 和 [开启知识图谱功能](../核心功能/开启知识图谱功能.md) diff --git a/docs/开发指南.md b/docs/开发指南.md index ddc9530f..5d7a60b1 100644 --- a/docs/开发指南.md +++ b/docs/开发指南.md @@ -18,7 +18,6 @@ make dev-start - MinIO（对象存储） - Neo4j（图数据库） - DocReader（文档读取服务） -- Jaeger（链路追踪） #### 2. 启动后端应用（新终端） @@ -79,7 +78,6 @@ make dev-stop - **Redis**: localhost:6379 - **MinIO Console**: http://localhost:9001 - **Neo4j Browser**: http://localhost:7474 -- **Jaeger UI**: http://localhost:16686 ## 开发工作流对比 @@ -227,7 +225,6 @@ VS Code 配置示例（`.vscode/launch.json`）： "DOCREADER_ADDR": "localhost:50051", "MINIO_ENDPOINT": "localhost:9000", "REDIS_ADDR": "localhost:6379", - "OTEL_EXPORTER_OTLP_ENDPOINT": "localhost:4317", "NEO4J_URI": "bolt://localhost:7687" }, "args": [] diff --git a/go.mod b/go.mod index 4655608f..341df9b2 100644 --- a/go.mod +++ b/go.mod @@ -65,12 +65,6 @@ require ( github.com/weaviate/weaviate-go-client/v5 v5.7.3 github.com/xuri/excelize/v2 v2.10.1 github.com/yanyiwu/gojieba v1.4.7 - go.opentelemetry.io/otel v1.43.0 - go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 - go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0 - go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0 - go.opentelemetry.io/otel/sdk v1.43.0 - go.opentelemetry.io/otel/trace v1.43.0 go.uber.org/dig v1.19.0 golang.org/x/crypto v0.51.0 golang.org/x/mod v0.36.0 @@ -310,7 +304,12 @@ require ( go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect + go.opentelemetry.io/otel v1.43.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0 // indirect go.opentelemetry.io/otel/metric v1.43.0 // indirect + go.opentelemetry.io/otel/sdk v1.43.0 // indirect + go.opentelemetry.io/otel/trace v1.43.0 // indirect go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/automaxprocs v1.5.3 // indirect diff --git a/go.sum b/go.sum index 0f7fee62..b6f59e85 100644 --- a/go.sum +++ b/go.sum @@ -2669,8 +2669,6 @@ go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0/go.mod h1:Vl1/iaggsuRl go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.0.1/go.mod h1:xOvWoTOrQjxjW61xtOmD/WKGRYb/P4NzRo3bs65U6Rk= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0 h1:RAE+JPfvEmvy+0LzyUA25/SGawPwIUbZ6u0Wug54sLc= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0/go.mod h1:AGmbycVGEsRx9mXMZ75CsOyhSP6MFIcj/6dnG+vhVjk= -go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0 h1:mS47AX77OtFfKG4vtp+84kuGSFZHTyxtXIN269vChY0= -go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0/go.mod h1:PJnsC41lAGncJlPUniSwM81gc80GkgWJWr3cu2nKEtU= go.opentelemetry.io/otel/metric v1.19.0/go.mod h1:L5rUsV9kM1IxCj1MmSdS+JQAcVm319EUrDVLrt7jqt8= go.opentelemetry.io/otel/metric v1.21.0/go.mod h1:o1p3CA8nNHW8j5yuQLdc1eeqEaPfzug24uvsyIEJRWM= go.opentelemetry.io/otel/metric v1.22.0/go.mod h1:evJGjVpZv0mQ5QBRJoBF64yMuOf4xCWdXjK8pzFvliY= diff --git a/helm/README.md b/helm/README.md index 260d74a3..e685e49c 100644 --- a/helm/README.md +++ b/helm/README.md @@ -230,7 +230,6 @@ These map to docker-compose profiles: | `minio.enabled` | Enable MinIO storage | `false` | | `neo4j.enabled` | Enable Neo4j (GraphRAG) | `false` | | `qdrant.enabled` | Enable Qdrant vector DB | `false` | -| `jaeger.enabled` | Enable Jaeger tracing | `false` | ## Security Best Practices diff --git a/helm/templates/NOTES.txt b/helm/templates/NOTES.txt index c7dc1bf5..4c4030b7 100644 --- a/helm/templates/NOTES.txt +++ b/helm/templates/NOTES.txt @@ -65,7 +65,7 @@ Components deployed: - Redis ({{ include "weknora.redis.image" . }}) {{- end }} -{{- if or .Values.minio.enabled .Values.neo4j.enabled .Values.qdrant.enabled .Values.jaeger.enabled }} +{{- if or .Values.minio.enabled .Values.neo4j.enabled .Values.qdrant.enabled }} Optional components enabled: {{- if .Values.minio.enabled }} @@ -77,9 +77,6 @@ Optional components enabled: {{- if .Values.qdrant.enabled }} - Qdrant (Vector Database) {{- end }} -{{- if .Values.jaeger.enabled }} - - Jaeger (Distributed Tracing) -{{- end }} {{- end }} -------------------------------------------------------------------------------- diff --git a/helm/values.yaml b/helm/values.yaml index f0504a48..3f119e87 100644 --- a/helm/values.yaml +++ b/helm/values.yaml @@ -492,12 +492,3 @@ qdrant: persistence: enabled: true size: 10Gi - -# -- Jaeger configuration (Distributed Tracing) -# Equivalent to: docker compose --profile jaeger -jaeger: - # -- Enable Jaeger tracing - enabled: false - image: - repository: jaegertracing/all-in-one - tag: latest diff --git a/internal/application/service/knowledge_process.go b/internal/application/service/knowledge_process.go index 9c65bb84..e85d65b7 100644 --- a/internal/application/service/knowledge_process.go +++ b/internal/application/service/knowledge_process.go @@ -19,14 +19,12 @@ import ( "github.com/Tencent/WeKnora/internal/models/chat" "github.com/Tencent/WeKnora/internal/models/embedding" "github.com/Tencent/WeKnora/internal/searchutil" - "github.com/Tencent/WeKnora/internal/tracing" "github.com/Tencent/WeKnora/internal/tracing/langfuse" "github.com/Tencent/WeKnora/internal/types" "github.com/Tencent/WeKnora/internal/types/interfaces" secutils "github.com/Tencent/WeKnora/internal/utils" "github.com/google/uuid" "github.com/hibiken/asynq" - "go.opentelemetry.io/otel/attribute" ) func (s *knowledgeService) cloneKnowledge( @@ -258,22 +256,11 @@ func (s *knowledgeService) processChunks(ctx context.Context, options = opts[0] } - ctx, span := tracing.ContextWithSpan(ctx, "knowledgeService.processChunks") - defer span.End() - span.SetAttributes( - attribute.Int("tenant_id", int(knowledge.TenantID)), - attribute.String("knowledge_base_id", knowledge.KnowledgeBaseID), - attribute.String("knowledge_id", knowledge.ID), - attribute.String("embedding_model_id", kb.EmbeddingModelID), - attribute.Int("chunk_count", len(chunks)), - ) - // Check if knowledge is being deleted/cancelled before processing. // Both statuses short-circuit identically here — there's nothing to clean // up yet so the branch is purely "stop early". if aborted, status := s.isKnowledgeAborted(ctx, knowledge.TenantID, knowledge.ID); aborted { logger.Infof(ctx, "Knowledge aborted (%s), skipping chunk processing: %s", status, knowledge.ID) - span.AddEvent("aborted: knowledge " + status) return } @@ -284,7 +271,6 @@ func (s *knowledgeService) processChunks(ctx context.Context, embeddingModel, err = s.modelService.GetEmbeddingModel(ctx, kb.EmbeddingModelID) if err != nil { logger.GetLogger(ctx).WithField("error", err).Errorf("processChunks get embedding model failed") - span.RecordError(err) return } } else { @@ -489,14 +475,12 @@ func (s *knowledgeService) processChunks(ctx context.Context, // Nothing has been persisted yet, so both branches just bail. if aborted, status := s.isKnowledgeAborted(ctx, knowledge.TenantID, knowledge.ID); aborted { logger.Infof(ctx, "Knowledge aborted (%s), skipping chunk write: %s", status, knowledge.ID) - span.AddEvent("aborted: knowledge " + status + " before saving") return } // Save chunks to database — ALWAYS, regardless of indexing strategy. // Chunks are needed for wiki generation, graph extraction, and summary generation // even when vector/keyword indexing is disabled. - span.AddEvent("create chunks") s.beginStage(ctx, knowledge.ID, types.StageChunking, types.JSONMap{ "chunks_planned": len(insertChunks), }) @@ -505,7 +489,6 @@ func (s *knowledgeService) processChunks(ctx context.Context, knowledge.ErrorMessage = err.Error() knowledge.UpdatedAt = time.Now() s.repo.UpdateKnowledge(ctx, knowledge) - span.RecordError(err) s.failStage(ctx, knowledge.ID, types.StageChunking, werrors.ErrCodeChunkingFailed, "create chunks failed", err) return @@ -557,7 +540,6 @@ func (s *knowledgeService) processChunks(ctx context.Context, } // Calculate storage size required for embeddings - span.AddEvent("estimate storage size") totalStorageSize = retrieveEngine.EstimateStorageSize(ctx, embeddingModel, indexInfoList) if tenantInfo.StorageQuota > 0 { // Re-fetch tenant storage information @@ -567,7 +549,6 @@ func (s *knowledgeService) processChunks(ctx context.Context, knowledge.ErrorMessage = err.Error() knowledge.UpdatedAt = time.Now() s.repo.UpdateKnowledge(ctx, knowledge) - span.RecordError(err) return } // Check if there's enough storage quota available @@ -576,7 +557,6 @@ func (s *knowledgeService) processChunks(ctx context.Context, knowledge.ErrorMessage = "存储空间不足" knowledge.UpdatedAt = time.Now() s.repo.UpdateKnowledge(ctx, knowledge) - span.RecordError(errors.New("storage quota exceeded")) return } } @@ -591,11 +571,9 @@ func (s *knowledgeService) processChunks(ctx context.Context, logger.Warnf(ctx, "Failed to cleanup chunks after deletion detected: %v", err) } } - span.AddEvent("aborted: knowledge " + status + " before indexing") return } - span.AddEvent("batch index") err = retrieveEngine.BatchIndex(ctx, embeddingModel, indexInfoList) if err != nil { knowledge.ParseStatus = types.ParseStatusFailed @@ -614,7 +592,6 @@ func (s *knowledgeService) processChunks(ctx context.Context, ); err != nil { logger.Errorf(ctx, "Delete index failed: %v", err) } - span.RecordError(err) // Map vector store / embedding rate-limit errors to a // stable code so the UI can offer "retry later" hints. code := werrors.ErrCodeVectorStoreWriteFailed @@ -645,7 +622,6 @@ func (s *knowledgeService) processChunks(ctx context.Context, logger.Warnf(ctx, "Failed to cleanup index after deletion detected: %v", err) } } - span.AddEvent("aborted: knowledge " + status + " during processing") return } } else { diff --git a/internal/application/service/retriever/composite.go b/internal/application/service/retriever/composite.go index 6173d216..977dd049 100644 --- a/internal/application/service/retriever/composite.go +++ b/internal/application/service/retriever/composite.go @@ -11,10 +11,8 @@ import ( "github.com/Tencent/WeKnora/internal/common" "github.com/Tencent/WeKnora/internal/logger" "github.com/Tencent/WeKnora/internal/models/embedding" - "github.com/Tencent/WeKnora/internal/tracing" "github.com/Tencent/WeKnora/internal/types" "github.com/Tencent/WeKnora/internal/types/interfaces" - "go.opentelemetry.io/otel/attribute" ) // engineInfo holds information about a retrieve engine and its supported retriever types @@ -200,8 +198,6 @@ func (c *CompositeRetrieveEngine) concurrentExecWithError( func (c *CompositeRetrieveEngine) Index(ctx context.Context, embedder embedding.Embedder, indexInfo *types.IndexInfo, ) error { - ctx, span := tracing.ContextWithSpan(ctx, "CompositeRetrieveEngine.Index") - defer span.End() err := c.concurrentExecWithError(ctx, func(ctx context.Context, engineInfo *engineInfo) error { if err := engineInfo.retrieveEngine.Index(ctx, embedder, indexInfo, engineInfo.retrieverType); err != nil { logger.Errorf(ctx, "Repository %s failed to save: %v", engineInfo.retrieveEngine.EngineType(), err) @@ -209,11 +205,6 @@ func (c *CompositeRetrieveEngine) Index(ctx context.Context, } return nil }) - span.RecordError(err) - span.SetAttributes( - attribute.String("embedder", embedder.GetModelName()), - attribute.String("source_id", indexInfo.SourceID), - ) return err } @@ -221,8 +212,6 @@ func (c *CompositeRetrieveEngine) Index(ctx context.Context, func (c *CompositeRetrieveEngine) BatchIndex(ctx context.Context, embedder embedding.Embedder, indexInfoList []*types.IndexInfo, ) error { - ctx, span := tracing.ContextWithSpan(ctx, "CompositeRetrieveEngine.BatchIndex") - defer span.End() // Deduplicate sourceIDs indexInfoList = common.Deduplicate(func(info *types.IndexInfo) string { return info.SourceID }, indexInfoList...) err := c.concurrentExecWithError(ctx, func(ctx context.Context, engineInfo *engineInfo) error { @@ -237,11 +226,6 @@ func (c *CompositeRetrieveEngine) BatchIndex(ctx context.Context, } return nil }) - span.RecordError(err) - span.SetAttributes( - attribute.String("embedder", embedder.GetModelName()), - attribute.Int("index_info_count", len(indexInfoList)), - ) return err } @@ -318,18 +302,13 @@ func (c *CompositeRetrieveEngine) DeleteByKnowledgeIDList(ctx context.Context, func (c *CompositeRetrieveEngine) EstimateStorageSize(ctx context.Context, embedder embedding.Embedder, indexInfoList []*types.IndexInfo, ) int64 { - ctx, span := tracing.ContextWithSpan(ctx, "CompositeRetrieveEngine.EstimateStorageSize") - defer span.End() sum := atomic.Int64{} err := c.concurrentExecWithError(ctx, func(ctx context.Context, engineInfo *engineInfo) error { sum.Add(engineInfo.retrieveEngine.EstimateStorageSize(ctx, embedder, indexInfoList, engineInfo.retrieverType)) return nil }) - span.RecordError(err) - span.SetAttributes( - attribute.String("embedder", embedder.GetModelName()), - attribute.Int("index_info_count", len(indexInfoList)), - attribute.Int64("storage_size", sum.Load()), - ) + if err != nil { + logger.Errorf(ctx, "EstimateStorageSize failed: %v", err) + } return sum.Load() } diff --git a/internal/container/container.go b/internal/container/container.go index 9280ef0c..ebb49f3e 100644 --- a/internal/container/container.go +++ b/internal/container/container.go @@ -78,7 +78,6 @@ import ( "github.com/Tencent/WeKnora/internal/models/utils/ollama" "github.com/Tencent/WeKnora/internal/router" "github.com/Tencent/WeKnora/internal/stream" - "github.com/Tencent/WeKnora/internal/tracing" "github.com/Tencent/WeKnora/internal/tracing/langfuse" "github.com/Tencent/WeKnora/internal/types" "github.com/Tencent/WeKnora/internal/types/interfaces" @@ -107,15 +106,12 @@ func BuildContainer(container *dig.Container) *dig.Container { // Core infrastructure configuration logger.Debugf(ctx, "[Container] Registering core infrastructure...") must(container.Provide(config.LoadConfig)) - must(container.Provide(initTracer)) must(container.Provide(initLangfuse)) must(container.Provide(initDatabase)) must(container.Provide(initFileService)) must(container.Provide(initRedisClient)) must(container.Provide(initAntsPool)) - // Register tracer cleanup handler (tracer needs to be available for cleanup registration) - must(container.Invoke(registerTracerCleanup)) must(container.Invoke(registerLangfuseCleanup)) // Register goroutine pool cleanup handler @@ -414,18 +410,6 @@ func must(err error) { } } -// initTracer initializes OpenTelemetry tracer -// Sets up distributed tracing for observability across the application -// Parameters: -// - None -// -// Returns: -// - Configured tracer instance -// - Error if initialization fails -func initTracer() (*tracing.Tracer, error) { - return tracing.InitTracer() -} - // initLangfuse initializes the Langfuse ingestion client. // Configuration is read from LANGFUSE_* environment variables (see // docs/langfuse.md). Returns a disabled manager if credentials are absent — @@ -1205,19 +1189,6 @@ func registerPoolCleanup(pool *ants.Pool, cleaner interfaces.ResourceCleaner) { }) } -// registerTracerCleanup registers the tracer for cleanup -// Ensures proper cleanup of the tracer when application shuts down -// Parameters: -// - tracer: Tracer instance -// - cleaner: Resource cleaner -func registerTracerCleanup(tracer *tracing.Tracer, cleaner interfaces.ResourceCleaner) { - // Register the cleanup function - actual context will be provided during cleanup - cleaner.RegisterWithName("Tracer", func() error { - // Create context for cleanup with longer timeout for tracer shutdown - return tracer.Cleanup(context.Background()) - }) -} - // registerLangfuseCleanup ensures buffered Langfuse events are flushed on // shutdown. A 5-second timeout matches other external-service cleanups and // balances data durability against a slow remote endpoint holding up exit. diff --git a/internal/im/service.go b/internal/im/service.go index 8426b132..91eebaa8 100644 --- a/internal/im/service.go +++ b/internal/im/service.go @@ -20,14 +20,10 @@ import ( "github.com/Tencent/WeKnora/internal/event" "github.com/Tencent/WeKnora/internal/logger" "github.com/Tencent/WeKnora/internal/models/chat" - "github.com/Tencent/WeKnora/internal/tracing" "github.com/Tencent/WeKnora/internal/types" "github.com/Tencent/WeKnora/internal/types/interfaces" "github.com/google/uuid" "github.com/redis/go-redis/v9" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/codes" - "go.opentelemetry.io/otel/trace" "gorm.io/gorm" ) @@ -667,14 +663,6 @@ func (s *Service) LoadAndStartChannels() error { // the leader lock and opens the connection; other instances periodically // retry so they can take over if the leader dies. func (s *Service) StartChannel(channel *IMChannel) error { - _, span := tracing.ContextWithSpan(context.Background(), "im.StartChannel") - defer span.End() - span.SetAttributes( - attribute.String("im.channel_id", channel.ID), - attribute.String("im.platform", channel.Platform), - attribute.String("im.mode", channel.Mode), - ) - s.mu.Lock() factory, ok := s.adapterFactories[channel.Platform] if !ok { @@ -1036,18 +1024,6 @@ func (s *Service) isDuplicate(ctx context.Context, messageID string) bool { // HandleMessage processes an incoming IM message end-to-end using channel config. func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, channelID string) error { - ctx, span := tracing.ContextWithSpan(ctx, "im.HandleMessage") - defer span.End() - span.SetAttributes( - attribute.String("im.channel_id", channelID), - attribute.String("im.platform", string(msg.Platform)), - attribute.String("im.user_id", msg.UserID), - attribute.String("im.chat_id", msg.ChatID), - attribute.String("im.thread_id", msg.ThreadID), - attribute.String("im.message_type", string(msg.MessageType)), - attribute.Bool("im.has_quote", msg.Quote != nil), - ) - // Dedup: skip if this message was already processed (IM platforms may retry) if msg.MessageID != "" { if s.isDuplicate(ctx, msg.MessageID) { @@ -1081,8 +1057,6 @@ func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, chann } } - span.SetAttributes(attribute.String("im.session_mode", channel.SessionMode)) - // Resolve threadID for key building — only include in thread mode to avoid // leaking thread scope into user-mode rate limit / inflight keys. threadID := "" @@ -1223,7 +1197,6 @@ func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, chann pos, enqueueErr := s.qaQueue.Enqueue(req) if enqueueErr != nil { qaCancel() - span.AddEvent("queue rejected", trace.WithAttributes(attribute.String("reason", enqueueErr.Error()))) logger.Warnf(ctx, "[IM] Queue rejected: user=%s reason=%v", msg.UserID, enqueueErr) _ = adapter.SendReply(ctx, msg, &ReplyMessage{ Content: "当前排队人数较多，请稍后再试。", @@ -1254,13 +1227,7 @@ func (s *Service) HandleMessage(ctx context.Context, msg *IncomingMessage, chann // executeQARequest is the worker handler that runs the QA pipeline for a queued request. // It is called by qaQueue workers and must not block indefinitely. func (s *Service) executeQARequest(req *qaRequest) { - ctx, span := tracing.ContextWithSpan(req.ctx, "im.ExecuteQA") - defer span.End() - span.SetAttributes( - attribute.String("im.channel_id", req.channelID), - attribute.String("im.user_key", req.userKey), - attribute.String("im.user_id", req.msg.UserID), - ) + ctx := req.ctx defer req.cancel() // Track in-flight request so /stop can cancel it. @@ -1270,7 +1237,6 @@ func (s *Service) executeQARequest(req *qaRequest) { // Check if a pre-execution /stop was issued while this request was queued. if s.checkAndClearStopMarker(ctx, req.userKey) { - span.AddEvent("cancelled by remote /stop before execution") logger.Infof(ctx, "[IM] Request cancelled by remote /stop before execution: %s", req.userKey) return } @@ -1289,7 +1255,6 @@ func (s *Service) executeQARequest(req *qaRequest) { if !streamDisabled { if streamer, ok := req.adapter.(StreamSender); ok { if err := s.handleMessageStream(ctx, req.msg, req.session, req.agent, kbIDs, streamer, req.adapter, req.userKey, req.tenant); err != nil { - span.SetStatus(codes.Error, err.Error()) logger.Errorf(ctx, "[IM] Stream QA failed: %v", err) } return @@ -1299,7 +1264,6 @@ func (s *Service) executeQARequest(req *qaRequest) { // Non-streaming fallback: collect full answer then send. answer, err := s.runQA(ctx, req.session, req.msg.Content, req.agent, kbIDs, req.userKey, req.msg.Quote) if err != nil { - span.SetStatus(codes.Error, err.Error()) logger.Errorf(ctx, "[IM] QA failed: %v, sending fallback reply", err) answer = "抱歉，处理您的问题时出现了异常，请稍后再试。" } @@ -1329,14 +1293,6 @@ func (s *Service) handleCommand( channelSession *ChannelSession, customAgent *types.CustomAgent, ) error { - ctx, span := tracing.ContextWithSpan(ctx, "im.HandleCommand") - defer span.End() - span.SetAttributes( - attribute.String("im.command", cmd.Name()), - attribute.String("im.channel_id", channel.ID), - attribute.String("im.user_id", msg.UserID), - ) - agentName := "" if customAgent != nil { agentName = customAgent.Name diff --git a/internal/middleware/trace.go b/internal/middleware/trace.go deleted file mode 100644 index 9dbb2d85..00000000 --- a/internal/middleware/trace.go +++ /dev/null @@ -1,117 +0,0 @@ -package middleware - -import ( - "bytes" - "fmt" - "io" - "strings" - - "github.com/gin-gonic/gin" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/codes" - - "github.com/Tencent/WeKnora/internal/tracing" -) - -// Custom ResponseWriter to capture response content -type responseBodyWriter struct { - gin.ResponseWriter - body *bytes.Buffer -} - -// Override Write method to write response content to buffer and original writer -func (r responseBodyWriter) Write(b []byte) (int, error) { - r.body.Write(b) - return r.ResponseWriter.Write(b) -} - -// TracingMiddleware provides a Gin middleware that creates a trace span for each request -func TracingMiddleware() gin.HandlerFunc { - return func(c *gin.Context) { - // Extract trace context from request headers - propagator := tracing.GetTracer() - if propagator == nil { - c.Next() - return - } - - // Create new span - spanName := fmt.Sprintf("%s %s", c.Request.Method, c.FullPath()) - ctx, span := tracing.ContextWithSpan(c.Request.Context(), spanName) - defer span.End() - - // Set basic span attributes - span.SetAttributes( - attribute.String("http.method", c.Request.Method), - attribute.String("http.url", c.Request.URL.String()), - attribute.String("http.path", c.FullPath()), - ) - - // Record request headers (optional, or selectively record important headers) - for key, values := range c.Request.Header { - // Skip sensitive or unnecessary headers - if strings.ToLower(key) == "authorization" || strings.ToLower(key) == "cookie" { - continue - } - span.SetAttributes(attribute.String("http.request.header."+key, strings.Join(values, ";"))) - } - - // Record request body (for POST/PUT/PATCH requests) - if c.Request.Method == "POST" || c.Request.Method == "PUT" || c.Request.Method == "PATCH" { - if c.Request.Body != nil { - bodyBytes, _ := io.ReadAll(c.Request.Body) - span.SetAttributes(attribute.String("http.request.body", string(bodyBytes))) - // Reset request body because ReadAll consumes the Reader content - c.Request.Body = io.NopCloser(bytes.NewBuffer(bodyBytes)) - } - } - - // Record query parameters - if len(c.Request.URL.RawQuery) > 0 { - span.SetAttributes(attribute.String("http.request.query", c.Request.URL.RawQuery)) - } - - // Set request context with span context - c.Request = c.Request.WithContext(ctx) - - // Store tracing context in Gin context - c.Set("trace.span", span) - c.Set("trace.ctx", ctx) - - // Create response body capturer - responseBody := &bytes.Buffer{} - responseWriter := &responseBodyWriter{ - ResponseWriter: c.Writer, - body: responseBody, - } - c.Writer = responseWriter - - // Process request - c.Next() - - // Set response status code - statusCode := c.Writer.Status() - span.SetAttributes(attribute.Int("http.status_code", statusCode)) - - // Record response body - responseContent := responseBody.String() - if len(responseContent) > 0 { - span.SetAttributes(attribute.String("http.response.body", responseContent)) - } - - // Record response headers (optional, or selectively record important headers) - for key, values := range c.Writer.Header() { - span.SetAttributes(attribute.String("http.response.header."+key, strings.Join(values, ";"))) - } - - // Mark as error if status code >= 400 - if statusCode >= 400 { - span.SetStatus(codes.Error, fmt.Sprintf("HTTP %d", statusCode)) - if err := c.Errors.Last(); err != nil { - span.RecordError(err.Err) - } - } else { - span.SetStatus(codes.Ok, "") - } - } -} diff --git a/internal/router/router.go b/internal/router/router.go index 7107e3e4..d4e16eb9 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -141,9 +141,6 @@ func NewRouter(params RouterParams) *gin.Engine { // Diagnostic preview of presigned URLs (Admin only, behind auth middleware). servePresignedPreview(r, params.Config) - // 添加OpenTelemetry追踪中间件 - // r.Use(middleware.TracingMiddleware()) - // Langfuse observability — only active when LANGFUSE_* env vars are set. // The middleware is registered unconditionally; when disabled it's a no-op. r.Use(langfuse.GinMiddleware()) diff --git a/internal/tracing/init.go b/internal/tracing/init.go deleted file mode 100644 index 0d322459..00000000 --- a/internal/tracing/init.go +++ /dev/null @@ -1,105 +0,0 @@ -package tracing - -import ( - "context" - "log" - "os" - "time" - - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/exporters/otlp/otlptrace" - "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" - "go.opentelemetry.io/otel/exporters/stdout/stdouttrace" - "go.opentelemetry.io/otel/propagation" - "go.opentelemetry.io/otel/sdk/resource" - sdktrace "go.opentelemetry.io/otel/sdk/trace" - semconv "go.opentelemetry.io/otel/semconv/v1.24.0" - "go.opentelemetry.io/otel/trace" -) - -const ( - AppName = "WeKnoraApp" -) - -type Tracer struct { - Cleanup func(context.Context) error -} - -var tracer trace.Tracer - -// InitTracer initializes OpenTelemetry tracer -func InitTracer() (*Tracer, error) { - // Create resource description - labels := []attribute.KeyValue{ - semconv.TelemetrySDKLanguageGo, - semconv.ServiceNameKey.String(AppName), - } - res := resource.NewWithAttributes(semconv.SchemaURL, labels...) - var err error - - // First try to create OTLP exporter (can connect to Jaeger, Zipkin, etc.) - var traceExporter sdktrace.SpanExporter - if endpoint := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); endpoint != "" { - // Use gRPC exporter - client := otlptracegrpc.NewClient( - otlptracegrpc.WithEndpoint(endpoint), - otlptracegrpc.WithInsecure(), - ) - traceExporter, err = otlptrace.New(context.Background(), client) - if err != nil { - return nil, err - } - } else { - // If no OTLP endpoint is set, default to standard output - traceExporter, err = stdouttrace.New() - if err != nil { - return nil, err - } - } - - // Create batch SpanProcessor - bsp := sdktrace.NewBatchSpanProcessor(traceExporter) - - sampler := sdktrace.AlwaysSample() - - // Create and register TracerProvider - tp := sdktrace.NewTracerProvider( - sdktrace.WithSampler(sampler), - sdktrace.WithResource(res), - sdktrace.WithSpanProcessor(bsp), - ) - otel.SetTracerProvider(tp) - - // Set global propagator - otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator( - propagation.TraceContext{}, - propagation.Baggage{}, - )) - - // Create Tracer for project use - tracer = tp.Tracer(AppName) - - // Return cleanup function - return &Tracer{ - Cleanup: func(ctx context.Context) error { - ctx, cancel := context.WithTimeout(ctx, 5*time.Second) - defer cancel() - if err := tp.Shutdown(ctx); err != nil { - log.Printf("Error shutting down tracer provider: %v", err) - return err - } - return nil - }, - }, nil -} - -// GetTracer gets global Tracer -func GetTracer() trace.Tracer { - return tracer -} - -// Create context with span -func ContextWithSpan(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span) { - return GetTracer().Start(ctx, name, opts...) -} diff --git a/scripts/cloud-image/README.md b/scripts/cloud-image/README.md index 59802b0d..48d4fae2 100644 --- a/scripts/cloud-image/README.md +++ b/scripts/cloud-image/README.md @@ -88,7 +88,6 @@ WeKnora `docker-compose.yml` 大量服务是 **profile 限定**，本镜像只 | `minio` | 对象存储替代本地文件 | | `qdrant` / `milvus` / `weaviate` / `doris` | 替代 pgvector | | `neo4j` | GraphRAG 知识图谱 | -| `jaeger` | OpenTelemetry trace UI | | `langfuse` | 自建 Langfuse 可观测平台 | | `dex` | OIDC 登录 | | `odl-hybrid` | OpenDataLoader Docling hybrid（体积大，无预发布镜像，需 `--build`） | diff --git a/scripts/cloud-image/prepare.sh b/scripts/cloud-image/prepare.sh index 485b36e0..18c4f0f6 100755 --- a/scripts/cloud-image/prepare.sh +++ b/scripts/cloud-image/prepare.sh @@ -182,7 +182,7 @@ docker compose up -d echo "[prepare] 4.5/6 预拉 sandbox 镜像 (Agent Skills 用, 非常驻)" docker compose --profile full pull sandbox || true -# 其他向量库 / 可观测组件 (qdrant, milvus, weaviate, doris, neo4j, langfuse-*, minio, jaeger, dex) +# 其他向量库 / 可观测组件 (qdrant, milvus, weaviate, doris, neo4j, langfuse-*, minio, dex) # 不预拉, 体积可省 5-15GB. 用户如需启用: # cd /opt/WeKnora && docker compose --profile up -d diff --git a/scripts/dev.sh b/scripts/dev.sh index f435d53f..6665bb6e 100755 --- a/scripts/dev.sh +++ b/scripts/dev.sh @@ -68,7 +68,6 @@ show_help() { echo " --minio 启动 MinIO 对象存储" echo " --qdrant 启动 Qdrant 向量数据库" echo " --neo4j 启动 Neo4j 图数据库" - echo " --jaeger 启动 Jaeger 链路追踪" echo " --dex 启动 Dex（OIDC 身份认证）" echo " --langfuse 启动 Langfuse（默认已开启）" echo " --no-langfuse 不启动 Langfuse" @@ -78,7 +77,6 @@ show_help() { echo "示例：" echo " $0 start # 启动基础服务" echo " $0 start --qdrant # 启动基础服务 + Qdrant" - echo " $0 start --qdrant --jaeger # 启动基础服务 + Qdrant + Jaeger" echo " $0 start --dex # 启动基础服务 + Dex" echo " $0 start --odl-hybrid # 启动基础服务 + OpenDataLoader hybrid" echo " $0 start --full # 启动所有服务" @@ -172,7 +170,7 @@ start_services() { # 解析 profile 参数 shift # 移除 "start" 命令本身 # 默认启动基础设施（postgres / redis / docreader）+ langfuse， - # 其余可选服务通过 --minio / --qdrant / --neo4j / --jaeger / --dex / --full 按需开启。 + # 其余可选服务通过 --minio / --qdrant / --neo4j / --dex / --full 按需开启。 PROFILES="--profile langfuse" ENABLED_SERVICES="langfuse" while [ $# -gt 0 ]; do @@ -189,10 +187,6 @@ start_services() { PROFILES="$PROFILES --profile neo4j" ENABLED_SERVICES="$ENABLED_SERVICES neo4j" ;; - --jaeger) - PROFILES="$PROFILES --profile jaeger" - ENABLED_SERVICES="$ENABLED_SERVICES jaeger" - ;; --dex) PROFILES="$PROFILES --profile dex" ENABLED_SERVICES="$ENABLED_SERVICES dex" @@ -212,7 +206,7 @@ start_services() { ;; --full) PROFILES="--profile full" - ENABLED_SERVICES="minio qdrant neo4j jaeger dex" + ENABLED_SERVICES="minio qdrant neo4j dex" break ;; *) @@ -254,9 +248,6 @@ start_services() { if [[ "$ENABLED_SERVICES" == *"neo4j"* ]]; then echo " - Neo4j: localhost:7474 (Bolt: localhost:7687)" fi - if [[ "$ENABLED_SERVICES" == *"jaeger"* ]]; then - echo " - Jaeger: localhost:16686" - fi if [[ "$ENABLED_SERVICES" == *"dex"* ]]; then echo " - Dex: localhost:5556" fi @@ -349,7 +340,6 @@ start_app() { export MINIO_ENDPOINT=localhost:9000 export REDIS_ADDR=localhost:6379 export MILVUS_ADDRESS=localhost:19530 - export OTEL_EXPORTER_OTLP_ENDPOINT=localhost:4317 export NEO4J_URI=bolt://localhost:7687 export QDRANT_HOST=localhost diff --git a/scripts/quick-dev.sh b/scripts/quick-dev.sh index 93aebd3c..f7f56eef 100755 --- a/scripts/quick-dev.sh +++ b/scripts/quick-dev.sh @@ -99,7 +99,6 @@ log_info "访问地址:" echo " - 前端: http://localhost:5173" echo " - 后端 API: http://localhost:8080" echo " - MinIO Console: http://localhost:9001" -echo " - Jaeger UI: http://localhost:16686" echo "" log_info "管理命令:" diff --git a/scripts/start_all.sh b/scripts/start_all.sh index 22552a9e..42a78b19 100755 --- a/scripts/start_all.sh +++ b/scripts/start_all.sh @@ -749,7 +749,6 @@ else log_success "所有服务启动完成，可通过以下地址访问:" printf "%b\n" "${GREEN} - 前端界面: http://localhost:${FRONTEND_PORT:-80}${NC}" printf "%b\n" "${GREEN} - API接口: http://localhost:${APP_PORT:-8080}${NC}" - printf "%b\n" "${GREEN} - Jaeger链路追踪: http://localhost:16686${NC}" echo "" log_info "正在持续输出容器日志（按 Ctrl+C 退出日志，容器不会停止）..." "$DOCKER_COMPOSE_BIN" $DOCKER_COMPOSE_SUBCMD logs app docreader postgres --since=10s -f @@ -763,7 +762,6 @@ else log_success "Docker容器启动完成，可通过以下地址访问:" printf "%b\n" "${GREEN} - 前端界面: http://localhost:${FRONTEND_PORT:-80}${NC}" printf "%b\n" "${GREEN} - API接口: http://localhost:${APP_PORT:-8080}${NC}" - printf "%b\n" "${GREEN} - Jaeger链路追踪: http://localhost:16686${NC}" echo "" log_info "正在持续输出容器日志（按 Ctrl+C 退出日志，容器不会停止）..." "$DOCKER_COMPOSE_BIN" $DOCKER_COMPOSE_SUBCMD logs app docreader postgres --since=10s -f