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