From c9539748a35fe7da84412c5f64c137991d2b3916 Mon Sep 17 00:00:00 2001 From: yokowu <18836617@qq.com> Date: Thu, 16 Jul 2026 18:51:23 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E5=96=84=E5=85=A8=E9=93=BE?= =?UTF-8?q?=E8=B7=AF=E8=BF=BD=E8=B8=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/biz/host/handler/v1/internal.go | 46 +++-- backend/cmd/server/main.go | 15 +- backend/config/config.go | 16 +- backend/go.mod | 20 +- backend/go.sum | 28 ++- backend/pkg/logger/context.go | 3 + backend/pkg/logger/logger.go | 2 +- backend/pkg/register.go | 3 + backend/pkg/request/request.go | 10 +- backend/pkg/request/request_trace_test.go | 43 ++++ backend/pkg/taskflow/filemanage.go | 5 +- backend/pkg/taskflow/task.go | 24 +++ backend/pkg/taskflow/task_live.go | 5 +- backend/pkg/taskflow/vm.go | 5 +- backend/pkg/telemetry/context.go | 89 +++++++++ backend/pkg/telemetry/http.go | 29 +++ backend/pkg/telemetry/http_test.go | 65 ++++++ backend/pkg/telemetry/sanitize.go | 128 ++++++++++++ backend/pkg/telemetry/sanitize_test.go | 77 +++++++ backend/pkg/telemetry/telemetry.go | 137 +++++++++++++ .../adr/0001-backend-distributed-tracing.md | 189 ++++++++++++++++++ docs/architecture/observability-glossary.md | 61 ++++++ 22 files changed, 962 insertions(+), 38 deletions(-) create mode 100644 backend/pkg/request/request_trace_test.go create mode 100644 backend/pkg/telemetry/context.go create mode 100644 backend/pkg/telemetry/http.go create mode 100644 backend/pkg/telemetry/http_test.go create mode 100644 backend/pkg/telemetry/sanitize.go create mode 100644 backend/pkg/telemetry/sanitize_test.go create mode 100644 backend/pkg/telemetry/telemetry.go create mode 100644 docs/architecture/adr/0001-backend-distributed-tracing.md create mode 100644 docs/architecture/observability-glossary.md diff --git a/backend/biz/host/handler/v1/internal.go b/backend/biz/host/handler/v1/internal.go index 115182148..dd1e0794e 100644 --- a/backend/biz/host/handler/v1/internal.go +++ b/backend/biz/host/handler/v1/internal.go @@ -26,6 +26,7 @@ import ( "github.com/chaitin/MonkeyCode/backend/pkg/entx" "github.com/chaitin/MonkeyCode/backend/pkg/lifecycle" "github.com/chaitin/MonkeyCode/backend/pkg/taskflow" + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" "github.com/chaitin/MonkeyCode/backend/pkg/ws" ) @@ -115,8 +116,9 @@ type VMActivityReq struct { // ReportHostInfo 上报宿主机信息 func (h *InternalHostHandler) ReportHostInfo(c *web.Context, host taskflow.Host) error { - if err := h.repo.UpsertHost(context.Background(), &host); err != nil { - h.logger.ErrorContext(context.Background(), "upsert host failed", "error", err) + ctx := c.Request().Context() + if err := h.repo.UpsertHost(ctx, &host); err != nil { + h.logger.ErrorContext(ctx, "upsert host failed", "error", err) return err } return c.Success(nil) @@ -124,8 +126,9 @@ func (h *InternalHostHandler) ReportHostInfo(c *web.Context, host taskflow.Host) // ReportVirtualMachine 上报虚拟机信息 func (h *InternalHostHandler) ReportVirtualMachine(c *web.Context, vm taskflow.VirtualMachine) error { - if err := h.repo.UpsertVirtualMachine(context.Background(), &vm); err != nil { - h.logger.ErrorContext(context.Background(), "upsert virtual machine failed", "error", err) + ctx := telemetry.WithVMID(c.Request().Context(), vm.ID) + if err := h.repo.UpsertVirtualMachine(ctx, &vm); err != nil { + h.logger.ErrorContext(ctx, "upsert virtual machine failed", "error", err) return err } return c.Success(nil) @@ -139,7 +142,7 @@ func (h *InternalHostHandler) VMActivity(c *web.Context, req VMActivityReq) erro return errors.New("last_active_at is required") } - ctx := c.Request().Context() + ctx := telemetry.WithVMID(c.Request().Context(), req.VMID) activeAt := time.Unix(req.LastActiveAt, 0) if now := time.Now(); activeAt.After(now) { h.logger.WarnContext(ctx, "vm activity timestamp is in the future", "vm_id", req.VMID, "last_active_at", req.LastActiveAt) @@ -162,14 +165,15 @@ func (h *InternalHostHandler) VMActivity(c *web.Context, req VMActivityReq) erro // ListLLM 列出虚拟机关联的 LLM func (h *InternalHostHandler) ListLLM(c *web.Context, req taskflow.ListLLMReq) error { - vm, err := h.repo.GetVirtualMachine(c.Request().Context(), req.VmID) + ctx := telemetry.WithVMID(c.Request().Context(), req.VmID) + vm, err := h.repo.GetVirtualMachine(ctx, req.VmID) if err != nil { - h.logger.ErrorContext(c.Request().Context(), "get virtual machine failed", "error", err) + h.logger.ErrorContext(ctx, "get virtual machine failed", "error", err) return err } if vm.HostID != req.HostID { - h.logger.ErrorContext(c.Request().Context(), "host id mismatch", "vm_host_id", vm.HostID, "req_host_id", req.HostID) + h.logger.ErrorContext(ctx, "host id mismatch", "vm_host_id", vm.HostID, "req_host_id", req.HostID) return errors.New("host id mismatch") } @@ -256,14 +260,19 @@ func (h *InternalHostHandler) CheckToken(c *web.Context, req taskflow.CheckToken h.logCheckTokenError(c.Request().Context(), logger, err) return err } + ctx := c.Request().Context() + ctx = telemetry.WithTaskID(ctx, tk.TaskID.String()) + ctx = telemetry.WithVMID(ctx, tk.Token) + ctx = telemetry.WithAgentSessionID(ctx, tk.SessionID) - logger.With("kind", tk.Kind, "vm_id", tk.Token).DebugContext(c.Request().Context(), "check token success") + logger.With("kind", tk.Kind, "vm_id", tk.Token).DebugContext(ctx, "check token success") return c.Success(tk) } func (h *InternalHostHandler) GetTaskLogStore(c *web.Context, req taskflow.GetTaskLogStoreReq) error { - store, err := h.taskRepo.GetLogStore(c.Request().Context(), req.TaskID) + ctx := telemetry.WithTaskID(c.Request().Context(), req.TaskID.String()) + store, err := h.taskRepo.GetLogStore(ctx, req.TaskID) if err != nil { return err } @@ -429,24 +438,26 @@ return nil // VmReady VM 就绪回调 func (h *InternalHostHandler) VmReady(c *web.Context, req taskflow.VirtualMachine) error { - h.logger.With("req", req).DebugContext(c.Request().Context(), "recv vm ready req") + ctx := telemetry.WithVMID(c.Request().Context(), req.ID) + h.logger.With("req", req).DebugContext(ctx, "recv vm ready req") - vm, err := h.repo.GetVirtualMachine(c.Request().Context(), req.ID) + vm, err := h.repo.GetVirtualMachine(ctx, req.ID) if err != nil { return err } for _, t := range vm.Edges.Tasks { - h.logger.With("task", t).DebugContext(c.Request().Context(), "vm-ready") + taskCtx := telemetry.WithTaskID(ctx, t.ID.String()) + h.logger.With("task", t).DebugContext(taskCtx, "vm-ready") if t.Status == consts.TaskStatusProcessing { continue } - if err := h.taskLifecycle.Transition(c.Request().Context(), t.ID, consts.TaskStatusProcessing, lifecycle.TaskMetadata{ + if err := h.taskLifecycle.Transition(taskCtx, t.ID, consts.TaskStatusProcessing, lifecycle.TaskMetadata{ TaskID: t.ID, UserID: t.UserID, }); err != nil { - h.logger.With("task", t, "error", err).ErrorContext(c.Request().Context(), "failed to transition task to processing") + h.logger.With("task", t, "error", err).ErrorContext(taskCtx, "failed to transition task to processing") } } @@ -504,7 +515,8 @@ func (h *InternalHostHandler) VmConditions(c *web.Context, req taskflow.VirtualM // GitCredential 获取 git 凭证 func (h *InternalHostHandler) GitCredential(c *web.Context, req taskflow.GitCredentialRequest) error { - ctx := c.Request().Context() + ctx := telemetry.WithTaskID(c.Request().Context(), req.TaskID) + ctx = telemetry.WithVMID(ctx, req.VMID) logger := h.logger.With("fn", "GitCredential", "task_id", req.TaskID, "vm_id", req.VMID) // 1. 有 task_id 时按任务链路取 token @@ -561,6 +573,8 @@ func (h *InternalHostHandler) GitCredential(c *web.Context, req taskflow.GitCred // GetTaskStreamIPs 获取任务 WebSocket 连接的客户端 IP func (h *InternalHostHandler) GetTaskStreamIPs(c *web.Context, req taskflow.GetTaskStreamIPsReq) error { + ctx := telemetry.WithTaskID(c.Request().Context(), req.TaskID) + c.SetRequest(c.Request().WithContext(ctx)) var ips []string if wsConn, ok := h.taskConns.Get(req.TaskID); ok { addr := wsConn.RemoteAddr() diff --git a/backend/cmd/server/main.go b/backend/cmd/server/main.go index e06b0c809..69e03dd08 100644 --- a/backend/cmd/server/main.go +++ b/backend/cmd/server/main.go @@ -15,6 +15,7 @@ import ( "github.com/chaitin/MonkeyCode/backend/pkg" "github.com/chaitin/MonkeyCode/backend/pkg/service" "github.com/chaitin/MonkeyCode/backend/pkg/store" + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" ) func main() { @@ -39,6 +40,19 @@ func main() { l := do.MustInvoke[*slog.Logger](injector) l.With("config", cfg).Debug("print config") + w := do.MustInvoke[*web.Web](injector) + shutdownTelemetry, err := telemetry.Setup(context.Background(), cfg.Telemetry) + if err != nil { + l.Warn("failed to setup telemetry, tracing disabled", "error", err) + shutdownTelemetry = func(context.Context) error { return nil } + } + defer func() { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := shutdownTelemetry(ctx); err != nil { + l.Warn("failed to shutdown telemetry", "error", err) + } + }() // 运行数据库迁移 if err := store.MigrateSQL(cfg, l); err != nil { @@ -53,7 +67,6 @@ func main() { biz.InvokeOpenSource(injector) // 获取 web 实例并启动服务 - w := do.MustInvoke[*web.Web](injector) w.PrintRoutes() svc := service.NewService( service.WithPprof(), diff --git a/backend/config/config.go b/backend/config/config.go index 2c4118804..1715cfa0c 100644 --- a/backend/config/config.go +++ b/backend/config/config.go @@ -9,6 +9,7 @@ import ( "github.com/chaitin/MonkeyCode/backend/consts" "github.com/chaitin/MonkeyCode/backend/pkg/logger" + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" ) type Config struct { @@ -41,10 +42,11 @@ type Config struct { RequestLogPath string `mapstructure:"request_log_path"` } `mapstructure:"llm_proxy"` - RootPath string `mapstructure:"root_path"` - Logger logger.Config `mapstructure:"logger"` - AdminToken string `mapstructure:"admin_token"` - Proxies []string `mapstructure:"proxies"` + RootPath string `mapstructure:"root_path"` + Logger logger.Config `mapstructure:"logger"` + Telemetry telemetry.Config `mapstructure:"telemetry"` + AdminToken string `mapstructure:"admin_token"` + Proxies []string `mapstructure:"proxies"` TaskFlow TaskFlow `mapstructure:"taskflow"` MCPHub MCPHub `mapstructure:"mcp_hub"` @@ -344,6 +346,12 @@ func Init(dir string) (*Config, error) { v.SetDefault("database.conn_max_lifetime", 30) v.SetDefault("root_path", "/app") v.SetDefault("logger.level", "info") + v.SetDefault("telemetry.enabled", false) + v.SetDefault("telemetry.endpoint", "") + v.SetDefault("telemetry.insecure", true) + v.SetDefault("telemetry.service_name", "monkeycode-backend") + v.SetDefault("telemetry.service_version", "") + v.SetDefault("telemetry.environment", "") v.SetDefault("session.expire_day", 30) v.SetDefault("smtp.host", "") v.SetDefault("smtp.port", 587) diff --git a/backend/go.mod b/backend/go.mod index a0b702d01..c2162c9e5 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -33,7 +33,12 @@ require ( github.com/samber/do v1.6.0 github.com/sashabaranov/go-openai v1.41.2 github.com/spf13/viper v1.21.0 - gitlab.com/gitlab-org/api/client-go v1.46.0 + gitlab.com/gitlab-org/api/client-go v1.8.1 + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 + go.opentelemetry.io/otel v1.42.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0 + go.opentelemetry.io/otel/sdk v1.42.0 + go.opentelemetry.io/otel/trace v1.42.0 golang.org/x/crypto v0.49.0 golang.org/x/oauth2 v0.36.0 golang.org/x/sync v0.20.0 @@ -66,8 +71,10 @@ require ( github.com/bmatcuk/doublestar v1.3.4 // indirect github.com/bradleyfalzon/ghinstallation/v2 v2.17.0 // indirect github.com/buger/jsonparser v1.1.2 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/gabriel-vasile/mimetype v1.4.13 // indirect github.com/go-faster/city v1.0.1 // indirect @@ -85,6 +92,7 @@ require ( github.com/google/go-github/v75 v75.0.0 // indirect github.com/google/go-querystring v1.2.0 // indirect github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect github.com/hashicorp/go-multierror v1.1.1 // indirect @@ -134,11 +142,9 @@ require ( github.com/zclconf/go-cty-yaml v1.1.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/github.com/labstack/echo/otelecho v0.67.0 // indirect - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 // indirect - go.opentelemetry.io/otel v1.42.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 // indirect go.opentelemetry.io/otel/metric v1.42.0 // indirect - go.opentelemetry.io/otel/sdk v1.42.0 // indirect - go.opentelemetry.io/otel/trace v1.42.0 // indirect + go.opentelemetry.io/proto/otlp v1.7.1 // indirect go.uber.org/atomic v1.11.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/mod v0.33.0 // indirect @@ -146,6 +152,10 @@ require ( golang.org/x/sys v0.42.0 // indirect golang.org/x/time v0.15.0 // indirect golang.org/x/tools v0.42.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 // indirect + google.golang.org/grpc v1.75.1 // indirect + google.golang.org/protobuf v1.36.10 // indirect gopkg.in/ini.v1 v1.66.2 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/backend/go.sum b/backend/go.sum index 67256edd9..be44769df 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -84,6 +84,8 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/buger/jsonparser v1.1.2 h1:frqHqw7otoVbk5M8LlE/L7HTnIq2v9RX6EJ48i9AxJk= github.com/buger/jsonparser v1.1.2/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58 h1:F1EaeKL/ta07PY/k9Os/UFtwERei2/XzGemhpGnBKNg= @@ -158,6 +160,8 @@ github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w github.com/golang-migrate/migrate/v4 v4.19.0 h1:RcjOnCGz3Or6HQYEJ/EEVLfWnmw9KnoigPSjzhCuaSE= github.com/golang-migrate/migrate/v4 v4.19.0/go.mod h1:9dyEcu+hO+G9hPSw8AIg50yg622pXJsoHItQnDGZkI0= github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= @@ -177,10 +181,10 @@ github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORR github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc= github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= -github.com/graph-gophers/graphql-go v1.9.0 h1:yu0ucKHLc5qGpRwLYKIWtr9bOoxovkWasuBrPQwlHls= -github.com/graph-gophers/graphql-go v1.9.0/go.mod h1:23olKZ7duEvHlF/2ELEoSZaY1aNPfShjP782SOoNTyM= github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 h1:+ngKgrYPPJrOjhax5N+uePQ0Fh1Z7PheYoUI/0nzkPA= github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnVTyacbefKhmbLhIhU= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= @@ -364,8 +368,8 @@ github.com/zclconf/go-cty-yaml v1.1.0 h1:nP+jp0qPHv2IhUVqmQSzjvqAWcObN0KBkUl2rWB github.com/zclconf/go-cty-yaml v1.1.0/go.mod h1:9YLUH4g7lOhVWqUbctnVlZ5KLpg7JAprQNgxSZ1Gyxs= github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0= github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA= -gitlab.com/gitlab-org/api/client-go v1.46.0 h1:YxBWFZIFYKcGESCb9fpkwzouo+apyB9pr/XTWzNoL24= -gitlab.com/gitlab-org/api/client-go v1.46.0/go.mod h1:FtgyU6g2HS5+fMhw6nLK96GBEEBx5MzntOiJWfIaiN8= +gitlab.com/gitlab-org/api/client-go v1.8.1 h1:YQyAh2Gd+NzcbRWWgDIi/pX0wLlm7QEZWtc0FikQRs4= +gitlab.com/gitlab-org/api/client-go v1.8.1/go.mod h1:tVIvZPcBPFPGYtLZOUIUafaZMmomCS0W81eACbn4Egw= go.mongodb.org/mongo-driver v1.11.4/go.mod h1:PTSz5yu21bkT/wXpkS7WR5f0ddqw5quethTUn9WM+2g= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= @@ -377,6 +381,10 @@ go.opentelemetry.io/contrib/propagators/b3 v1.42.0 h1:B2Pew5ufEtgkjLF+tSkXjgYZXQ go.opentelemetry.io/contrib/propagators/b3 v1.42.0/go.mod h1:iPgUcSEF5DORW6+yNbdw/YevUy+QqJ508ncjhrRSCjc= go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho= go.opentelemetry.io/otel v1.42.0/go.mod h1:lJNsdRMxCUIWuMlVJWzecSMuNjE7dOYyWlqOXWkdqCc= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 h1:GqRJVj7UmLjCVyVJ3ZFLdPRmhDUp2zFmQe3RHIOsw24= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0/go.mod h1:ri3aaHSmCTVYu2AWv44YMauwAQc0aqI9gHKIcSbI1pU= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0 h1:lwI4Dc5leUqENgGuQImwLo4WnuXFPetmPpkLi2IrX54= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0/go.mod h1:Kz/oCE7z5wuyhPxsXDuaPteSWqjSBD5YaSdbxZYGbGk= go.opentelemetry.io/otel/metric v1.42.0 h1:2jXG+3oZLNXEPfNmnpxKDeZsFI5o4J+nz6xUlaFdF/4= go.opentelemetry.io/otel/metric v1.42.0/go.mod h1:RlUN/7vTU7Ao/diDkEpQpnz3/92J9ko05BIwxYa2SSI= go.opentelemetry.io/otel/sdk v1.42.0 h1:LyC8+jqk6UJwdrI/8VydAq/hvkFKNHZVIWuslJXYsDo= @@ -385,6 +393,8 @@ go.opentelemetry.io/otel/sdk/metric v1.42.0 h1:D/1QR46Clz6ajyZ3G8SgNlTJKBdGp84q9 go.opentelemetry.io/otel/sdk/metric v1.42.0/go.mod h1:Ua6AAlDKdZ7tdvaQKfSmnFTdHx37+J4ba8MwVCYM5hc= go.opentelemetry.io/otel/trace v1.42.0 h1:OUCgIPt+mzOnaUTpOQcBiM/PLQ/Op7oq6g4LenLmOYY= go.opentelemetry.io/otel/trace v1.42.0/go.mod h1:f3K9S+IFqnumBkKhRJMeaZeNk9epyhnCmQh/EysQCdc= +go.opentelemetry.io/proto/otlp v1.7.1 h1:gTOMpGDb0WTBOP8JaO72iL3auEZhVmAQg4ipjOVAtj4= +go.opentelemetry.io/proto/otlp v1.7.1/go.mod h1:b2rVh6rfI/s2pHWNlB7ILJcRALpcNDzKhACevjI+ZnE= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= @@ -448,8 +458,18 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= +gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= +google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 h1:BIRfGDEjiHRrk0QKZe3Xv2ieMhtgRGeLcZQ0mIVn4EY= +google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5/go.mod h1:j3QtIyytwqGr1JUDtYXwtMXWPKsEa5LtzIFN1Wn5WvE= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 h1:eaY8u2EuxbRv7c3NiGK0/NedzVsCcV6hDuU5qPX5EGE= +google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5/go.mod h1:M4/wBTSeyLxupu3W3tJtOgB14jILAS/XWPSSa3TAlJc= +google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI= +google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= diff --git a/backend/pkg/logger/context.go b/backend/pkg/logger/context.go index 871047a44..25065d33d 100644 --- a/backend/pkg/logger/context.go +++ b/backend/pkg/logger/context.go @@ -3,6 +3,8 @@ package logger import ( "context" "log/slog" + + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" ) type contextKey string @@ -38,6 +40,7 @@ func (c *ContextLogger) Handle(ctx context.Context, r slog.Record) error { if i, ok := ctx.Value(UserIDKey).(string); ok { newRecord.AddAttrs(slog.String("user_id", i)) } + newRecord.AddAttrs(telemetry.LogAttrs(ctx)...) return c.Handler.Handle(ctx, newRecord) } diff --git a/backend/pkg/logger/logger.go b/backend/pkg/logger/logger.go index c43709f12..24cc93ab4 100644 --- a/backend/pkg/logger/logger.go +++ b/backend/pkg/logger/logger.go @@ -30,7 +30,7 @@ func NewLogger(cfg *Config) *slog.Logger { ReplaceAttr: replaceAttr, }) handler := &ContextLogger{Handler: base} - return slog.New(handler) + return slog.New(handler).With("service.name", "monkeycode-backend") } func replaceAttr(_ []string, attr slog.Attr) slog.Attr { diff --git a/backend/pkg/register.go b/backend/pkg/register.go index c1cbf92be..59dcbec52 100644 --- a/backend/pkg/register.go +++ b/backend/pkg/register.go @@ -34,6 +34,7 @@ import ( "github.com/chaitin/MonkeyCode/backend/pkg/tasker" "github.com/chaitin/MonkeyCode/backend/pkg/taskflow" "github.com/chaitin/MonkeyCode/backend/pkg/tasklog" + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" "github.com/chaitin/MonkeyCode/backend/pkg/vmrecycle" "github.com/chaitin/MonkeyCode/backend/pkg/ws" ) @@ -61,10 +62,12 @@ func RegisterInfra(i *do.Injector, w ...*web.Web) error { // Web if len(w) > 0 && w[0] != nil { + w[0].Echo().Pre(telemetry.SanitizeIncomingTrace) do.ProvideValue(i, w[0]) } else { do.Provide(i, func(i *do.Injector) (*web.Web, error) { w := web.New() + w.Echo().Pre(telemetry.SanitizeIncomingTrace) return w, nil }) } diff --git a/backend/pkg/request/request.go b/backend/pkg/request/request.go index bd9e7f52f..a403491ec 100644 --- a/backend/pkg/request/request.go +++ b/backend/pkg/request/request.go @@ -14,6 +14,7 @@ import ( "time" "github.com/google/uuid" + "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" ) // Client HTTP 客户端 @@ -53,9 +54,14 @@ func NewClient(scheme string, host string, timeout time.Duration, opts ...ReqOpt opt(req) } + transport := req.client.Transport if req.tr != nil { - req.client.Transport = req.tr + transport = req.tr } + if transport == nil { + transport = http.DefaultTransport + } + req.client.Transport = otelhttp.NewTransport(transport) return req } @@ -203,7 +209,7 @@ func PostURL[T any](ctx context.Context, rawURL string, body any, opts ...Opt) ( req.Header.Set(k, v) } - resp, err := http.DefaultClient.Do(req) + resp, err := otelhttp.DefaultClient.Do(req) if err != nil { return nil, err } diff --git a/backend/pkg/request/request_trace_test.go b/backend/pkg/request/request_trace_test.go new file mode 100644 index 000000000..e2d8ad3eb --- /dev/null +++ b/backend/pkg/request/request_trace_test.go @@ -0,0 +1,43 @@ +package request + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/propagation" + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +func TestClientPropagatesTraceContext(t *testing.T) { + var traceparent string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + traceparent = r.Header.Get("traceparent") + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer server.Close() + + tp := sdktrace.NewTracerProvider(sdktrace.WithSampler(sdktrace.AlwaysSample())) + otel.SetTracerProvider(tp) + otel.SetTextMapPropagator(propagation.TraceContext{}) + defer func() { _ = tp.Shutdown(context.Background()) }() + + u, err := url.Parse(server.URL) + if err != nil { + t.Fatal(err) + } + client := NewClient(u.Scheme, u.Host, 5*time.Second) + ctx, span := tp.Tracer("test").Start(context.Background(), "request") + defer span.End() + if _, err := Get[map[string]bool](client, ctx, "/"); err != nil { + t.Fatal(err) + } + if traceparent == "" { + t.Fatal("traceparent is empty") + } +} diff --git a/backend/pkg/taskflow/filemanage.go b/backend/pkg/taskflow/filemanage.go index 4aca85a5c..26cb4dfa3 100644 --- a/backend/pkg/taskflow/filemanage.go +++ b/backend/pkg/taskflow/filemanage.go @@ -12,6 +12,7 @@ import ( "github.com/coder/websocket" "github.com/chaitin/MonkeyCode/backend/pkg/request" + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" ) type fileManageClient struct { @@ -57,7 +58,7 @@ func (f *fileManageClient) Download(ctx context.Context, req FileReq, fn func(ui values.Add("path", req.Path) u.RawQuery = values.Encode() - conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{}) + conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{HTTPHeader: telemetry.TraceHeader(ctx)}) if err != nil { return err } @@ -130,7 +131,7 @@ func (f *fileManageClient) Upload(ctx context.Context, req FileReq, data <-chan values.Add("path", req.Path) u.RawQuery = values.Encode() - conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{}) + conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{HTTPHeader: telemetry.TraceHeader(ctx)}) if err != nil { return err } diff --git a/backend/pkg/taskflow/task.go b/backend/pkg/taskflow/task.go index 798e72497..3ecbfb9ac 100644 --- a/backend/pkg/taskflow/task.go +++ b/backend/pkg/taskflow/task.go @@ -4,6 +4,7 @@ import ( "context" "github.com/chaitin/MonkeyCode/backend/pkg/request" + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" ) type taskClient struct { @@ -18,12 +19,18 @@ func newTaskClient(client *request.Client) TaskManager { // Create implements TaskManager. func (t *taskClient) Create(ctx context.Context, req CreateTaskReq) error { + ctx = telemetry.WithTaskID(ctx, req.ID.String()) + ctx = telemetry.WithVMID(ctx, req.VMID) + telemetry.MarkCritical(ctx) _, err := request.Post[Resp[any]](t.client, ctx, "/internal/task", req) return err } // Restart implements TaskManager. func (t *taskClient) Restart(ctx context.Context, req RestartTaskReq) (*RestartTaskResp, error) { + ctx = telemetry.WithTaskID(ctx, req.ID.String()) + ctx = telemetry.WithRequestID(ctx, req.RequestId) + telemetry.MarkCritical(ctx) resp, err := request.Post[Resp[*RestartTaskResp]](t.client, ctx, "/internal/task/restart", req) if err != nil { return nil, err @@ -33,22 +40,39 @@ func (t *taskClient) Restart(ctx context.Context, req RestartTaskReq) (*RestartT // Stop implements TaskManager. func (t *taskClient) Stop(ctx context.Context, req TaskReq) error { + ctx = taskContext(ctx, req) + telemetry.MarkCritical(ctx) _, err := request.Post[Resp[any]](t.client, ctx, "/internal/task/stop", req) return err } // Cancel implements TaskManager. func (t *taskClient) Cancel(ctx context.Context, req TaskReq) error { + ctx = taskContext(ctx, req) + telemetry.MarkCritical(ctx) + telemetry.SetOutcome(ctx, "cancelled") _, err := request.Post[Resp[any]](t.client, ctx, "/internal/task/cancel", req) return err } // Continue implements TaskManager. func (t *taskClient) Continue(ctx context.Context, req TaskReq) error { + ctx = taskContext(ctx, req) + telemetry.MarkCritical(ctx) _, err := request.Post[Resp[any]](t.client, ctx, "/internal/task/continue", req) return err } +func taskContext(ctx context.Context, req TaskReq) context.Context { + if req.Task != nil { + ctx = telemetry.WithTaskID(ctx, req.Task.ID.String()) + } + if req.VirtualMachine != nil { + ctx = telemetry.WithVMID(ctx, req.VirtualMachine.ID) + } + return ctx +} + // AutoApprove implements TaskManager. func (t *taskClient) AutoApprove(ctx context.Context, req TaskApproveReq) error { _, err := request.Post[Resp[any]](t.client, ctx, "/internal/task/auto-approve", req) diff --git a/backend/pkg/taskflow/task_live.go b/backend/pkg/taskflow/task_live.go index 360b3dd35..17f0f7c2a 100644 --- a/backend/pkg/taskflow/task_live.go +++ b/backend/pkg/taskflow/task_live.go @@ -7,6 +7,8 @@ import ( "net/url" "github.com/coder/websocket" + + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" ) // TaskLive 连接 taskflow 的 task-live WebSocket 并处理消息流 @@ -26,7 +28,8 @@ func (c *Client) TaskLive(ctx context.Context, taskID string, flush bool, fn fun values.Add("flush", fmt.Sprintf("%t", flush)) u.RawQuery = values.Encode() - conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{}) + ctx = telemetry.WithTaskID(ctx, taskID) + conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{HTTPHeader: telemetry.TraceHeader(ctx)}) if err != nil { return err } diff --git a/backend/pkg/taskflow/vm.go b/backend/pkg/taskflow/vm.go index 6f077a2bb..ab0f79d63 100644 --- a/backend/pkg/taskflow/vm.go +++ b/backend/pkg/taskflow/vm.go @@ -9,6 +9,7 @@ import ( "github.com/coder/websocket" "github.com/chaitin/MonkeyCode/backend/pkg/request" + "github.com/chaitin/MonkeyCode/backend/pkg/telemetry" ) type virtualMachineClient struct { @@ -100,7 +101,7 @@ func (v *virtualMachineClient) Terminal(ctx context.Context, req *TerminalReq) ( values.Add("mode", fmt.Sprintf("%d", req.Mode)) u.RawQuery = values.Encode() - conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{}) + conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{HTTPHeader: telemetry.TraceHeader(ctx)}) if err != nil { return nil, err } @@ -147,7 +148,7 @@ func (v *virtualMachineClient) Reports(ctx context.Context, req ReportSubscribeR } u.RawQuery = values.Encode() - conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{}) + conn, _, err := websocket.Dial(ctx, u.String(), &websocket.DialOptions{HTTPHeader: telemetry.TraceHeader(ctx)}) if err != nil { return nil, err } diff --git a/backend/pkg/telemetry/context.go b/backend/pkg/telemetry/context.go new file mode 100644 index 000000000..ca53f6a83 --- /dev/null +++ b/backend/pkg/telemetry/context.go @@ -0,0 +1,89 @@ +package telemetry + +import ( + "context" + "log/slog" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" +) + +type contextKey string + +const ( + taskIDKey contextKey = "monkeycode.task.id" + agentSessionIDKey contextKey = "monkeycode.agent.session.id" + businessReqIDKey contextKey = "monkeycode.request.id" + projectIDKey contextKey = "monkeycode.project.id" + vmIDKey contextKey = "taskflow.vm.id" + terminalIDKey contextKey = "taskflow.terminal.session.id" +) + +func WithTaskID(ctx context.Context, id string) context.Context { + return withID(ctx, taskIDKey, "monkeycode.task.id", id) +} + +func WithAgentSessionID(ctx context.Context, id string) context.Context { + return withID(ctx, agentSessionIDKey, "monkeycode.agent.session.id", id) +} + +func WithRequestID(ctx context.Context, id string) context.Context { + return withID(ctx, businessReqIDKey, "monkeycode.request.id", id) +} + +func WithProjectID(ctx context.Context, id string) context.Context { + return withID(ctx, projectIDKey, "monkeycode.project.id", id) +} + +func WithVMID(ctx context.Context, id string) context.Context { + return withID(ctx, vmIDKey, "taskflow.vm.id", id) +} + +func WithTerminalSessionID(ctx context.Context, id string) context.Context { + return withID(ctx, terminalIDKey, "taskflow.terminal.session.id", id) +} + +func MarkCritical(ctx context.Context) { + trace.SpanFromContext(ctx).SetAttributes(attribute.String("telemetry.priority", "critical")) +} + +func SetOutcome(ctx context.Context, outcome string) { + if outcome != "" { + trace.SpanFromContext(ctx).SetAttributes(attribute.String("task.outcome", outcome)) + } +} + +func withID(ctx context.Context, key contextKey, name, id string) context.Context { + if id == "" { + return ctx + } + trace.SpanFromContext(ctx).SetAttributes(attribute.String(name, id)) + return context.WithValue(ctx, key, id) +} + +func LogAttrs(ctx context.Context) []slog.Attr { + attrs := make([]slog.Attr, 0, 8) + spanContext := trace.SpanContextFromContext(ctx) + if spanContext.IsValid() { + attrs = append(attrs, + slog.String("trace_id", spanContext.TraceID().String()), + slog.String("span_id", spanContext.SpanID().String()), + ) + } + for _, item := range []struct { + key contextKey + name string + }{ + {taskIDKey, "monkeycode.task.id"}, + {agentSessionIDKey, "monkeycode.agent.session.id"}, + {businessReqIDKey, "monkeycode.request.id"}, + {projectIDKey, "monkeycode.project.id"}, + {vmIDKey, "taskflow.vm.id"}, + {terminalIDKey, "taskflow.terminal.session.id"}, + } { + if value, ok := ctx.Value(item.key).(string); ok && value != "" { + attrs = append(attrs, slog.String(item.name, value)) + } + } + return attrs +} diff --git a/backend/pkg/telemetry/http.go b/backend/pkg/telemetry/http.go new file mode 100644 index 000000000..3a1013a81 --- /dev/null +++ b/backend/pkg/telemetry/http.go @@ -0,0 +1,29 @@ +package telemetry + +import ( + "context" + "net/http" + "strings" + + "github.com/labstack/echo/v4" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/propagation" +) + +func SanitizeIncomingTrace(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + header := c.Request().Header + header.Del("baggage") + if !strings.HasPrefix(c.Request().URL.Path, "/internal/") { + header.Del("traceparent") + header.Del("tracestate") + } + return next(c) + } +} + +func TraceHeader(ctx context.Context) http.Header { + header := make(http.Header) + otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(header)) + return header +} diff --git a/backend/pkg/telemetry/http_test.go b/backend/pkg/telemetry/http_test.go new file mode 100644 index 000000000..fb4a0c891 --- /dev/null +++ b/backend/pkg/telemetry/http_test.go @@ -0,0 +1,65 @@ +package telemetry + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/labstack/echo/v4" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/trace" +) + +func TestSanitizeIncomingTrace(t *testing.T) { + tests := []struct { + name string + path string + wantTraceparent bool + }{ + {name: "public", path: "/api/v1/tasks", wantTraceparent: false}, + {name: "internal", path: "/internal/task-report", wantTraceparent: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := echo.New() + req := httptest.NewRequest(http.MethodGet, tt.path, nil) + req.Header.Set("traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01") + req.Header.Set("tracestate", "vendor=value") + req.Header.Set("baggage", "secret=value") + ctx := e.NewContext(req, httptest.NewRecorder()) + + err := SanitizeIncomingTrace(func(c echo.Context) error { + if got := c.Request().Header.Get("traceparent"); (got != "") != tt.wantTraceparent { + t.Fatalf("traceparent = %q", got) + } + if got := c.Request().Header.Get("baggage"); got != "" { + t.Fatalf("baggage = %q", got) + } + return nil + })(ctx) + if err != nil { + t.Fatal(err) + } + }) + } +} + +func TestTraceHeader(t *testing.T) { + otel.SetTextMapPropagator(propagation.TraceContext{}) + spanContext := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: trace.TraceID{1}, + SpanID: trace.SpanID{2}, + TraceFlags: trace.FlagsSampled, + }) + ctx := trace.ContextWithSpanContext(context.Background(), spanContext) + header := TraceHeader(ctx) + if got := header.Get("traceparent"); got == "" { + t.Fatal("traceparent is empty") + } + if got := header.Get("baggage"); got != "" { + t.Fatalf("baggage = %q", got) + } +} diff --git a/backend/pkg/telemetry/sanitize.go b/backend/pkg/telemetry/sanitize.go new file mode 100644 index 000000000..2dc42728b --- /dev/null +++ b/backend/pkg/telemetry/sanitize.go @@ -0,0 +1,128 @@ +package telemetry + +import ( + "net/url" + + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +var allowedSpanAttributes = map[attribute.Key]struct{}{ + "monkeycode.task.id": {}, + "monkeycode.agent.session.id": {}, + "monkeycode.request.id": {}, + "monkeycode.project.id": {}, + "taskflow.vm.id": {}, + "taskflow.terminal.session.id": {}, + "taskflow.operation": {}, + "taskflow.stage": {}, + "telemetry.priority": {}, + "telemetry.noise": {}, + "task.outcome": {}, + "error.type": {}, + "exception.type": {}, + "http.request.method": {}, + "http.response.status_code": {}, + "http.route": {}, + "http.method": {}, + "http.status_code": {}, + "http.flavor": {}, + "http.scheme": {}, + "server.address": {}, + "server.port": {}, + "network.protocol.name": {}, + "network.protocol.version": {}, + "network.transport": {}, + "network.type": {}, + "net.host.name": {}, + "net.host.port": {}, + "net.protocol.name": {}, + "net.protocol.version": {}, + "net.transport": {}, + "rpc.system": {}, + "rpc.service": {}, + "rpc.method": {}, + "rpc.grpc.status_code": {}, + "db.system.name": {}, + "db.namespace": {}, + "db.operation.name": {}, + "db.collection.name": {}, + "db.system": {}, + "db.name": {}, + "db.operation": {}, + "db.sql.table": {}, +} + +func sanitizeAttributes(attrs []attribute.KeyValue) []attribute.KeyValue { + filtered := make([]attribute.KeyValue, 0, len(attrs)) + noise := false + hasNoise := false + for _, attr := range attrs { + if _, ok := allowedSpanAttributes[attr.Key]; ok { + filtered = append(filtered, attr) + if attr.Key == "telemetry.noise" { + hasNoise = true + if attr.Value.Type() == attribute.BOOL && attr.Value.AsBool() { + noise = true + } + } + continue + } + if isNoiseURL(attr) { + noise = true + } + } + if noise && !hasNoise { + filtered = append(filtered, attribute.Bool("telemetry.noise", true)) + } + return filtered +} + +func isNoiseURL(attr attribute.KeyValue) bool { + switch attr.Key { + case "url.full", "url.path", "http.url", "http.target": + default: + return false + } + if attr.Value.Type() != attribute.STRING { + return false + } + parsed, err := url.Parse(attr.Value.AsString()) + if err != nil { + return false + } + switch parsed.Path { + case "/health", "/metrics", "/internal/vm/activity", "/internal/stats", "/internal/host/is-online", "/internal/vm/is-online": + return true + default: + return false + } +} + +func sanitizeEvents(events []sdktrace.Event) []sdktrace.Event { + filtered := make([]sdktrace.Event, 0, len(events)) + for _, event := range events { + if event.Name != "exception" { + continue + } + event.Attributes = sanitizeAttributes(event.Attributes) + event.DroppedAttributeCount = 0 + filtered = append(filtered, event) + } + return filtered +} + +func sanitizeLinks(links []sdktrace.Link) []sdktrace.Link { + filtered := make([]sdktrace.Link, len(links)) + for i, link := range links { + link.Attributes = sanitizeAttributes(link.Attributes) + link.DroppedAttributeCount = 0 + filtered[i] = link + } + return filtered +} + +func sanitizeStatus(status sdktrace.Status) sdktrace.Status { + status.Description = "" + return status +} diff --git a/backend/pkg/telemetry/sanitize_test.go b/backend/pkg/telemetry/sanitize_test.go new file mode 100644 index 000000000..cf26296fb --- /dev/null +++ b/backend/pkg/telemetry/sanitize_test.go @@ -0,0 +1,77 @@ +package telemetry + +import ( + "testing" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +func TestSanitizeAttributesUsesAllowlist(t *testing.T) { + attrs := sanitizeAttributes([]attribute.KeyValue{ + attribute.String("monkeycode.task.id", "task-1"), + attribute.String("http.request.method", "POST"), + attribute.String("url.full", "https://example.com/internal/task?token=secret"), + attribute.String("http.request.body", "prompt-secret"), + attribute.String("db.query.text", "select * from secret"), + }) + got := make(map[string]string) + for _, attr := range attrs { + got[string(attr.Key)] = attr.Value.Emit() + } + if got["monkeycode.task.id"] != "task-1" || got["http.request.method"] != "POST" { + t.Fatalf("allowed attributes = %#v", got) + } + for _, key := range []string{"url.full", "http.request.body", "db.query.text"} { + if _, ok := got[key]; ok { + t.Fatalf("sensitive attribute %q was retained", key) + } + } +} + +func TestSanitizeAttributesMarksNoiseWithoutURL(t *testing.T) { + attrs := sanitizeAttributes([]attribute.KeyValue{ + attribute.String("url.full", "http://monkeycode/internal/vm/activity"), + }) + if len(attrs) != 1 || attrs[0].Key != "telemetry.noise" || !attrs[0].Value.AsBool() { + t.Fatalf("attributes = %#v", attrs) + } +} + +func TestSanitizeEventsAndStatusRemoveErrorDetails(t *testing.T) { + events := sanitizeEvents([]sdktrace.Event{{ + Name: "exception", + Attributes: []attribute.KeyValue{ + attribute.String("exception.type", "timeout"), + attribute.String("exception.message", "token=secret"), + attribute.String("exception.stacktrace", "secret stack"), + }, + }}) + if len(events) != 1 || len(events[0].Attributes) != 1 || events[0].Attributes[0].Key != "exception.type" { + t.Fatalf("events = %#v", events) + } + status := sanitizeStatus(sdktrace.Status{Code: codes.Error, Description: "token=secret"}) + if status.Code != codes.Error || status.Description != "" { + t.Fatalf("status = %#v", status) + } +} + +func TestBuildResourceUsesExplicitAllowlist(t *testing.T) { + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "secret.value=do-not-export") + res := buildResource(Config{ + ServiceName: "monkeycode-backend", + ServiceVersion: "v1", + Environment: "test", + }) + attrs := make(map[string]string) + for _, attr := range res.Attributes() { + attrs[string(attr.Key)] = attr.Value.AsString() + } + if attrs["service.name"] != "monkeycode-backend" || attrs["service.version"] != "v1" || attrs["deployment.environment.name"] != "test" { + t.Fatalf("resource attributes = %#v", attrs) + } + if _, ok := attrs["secret.value"]; ok { + t.Fatalf("resource attributes contain secret: %#v", attrs) + } +} diff --git a/backend/pkg/telemetry/telemetry.go b/backend/pkg/telemetry/telemetry.go new file mode 100644 index 000000000..b556b7888 --- /dev/null +++ b/backend/pkg/telemetry/telemetry.go @@ -0,0 +1,137 @@ +package telemetry + +import ( + "context" + "errors" + "os" + "strings" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc" + "go.opentelemetry.io/otel/propagation" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +type Config struct { + Enabled bool `mapstructure:"enabled"` + Endpoint string `mapstructure:"endpoint"` + Insecure bool `mapstructure:"insecure"` + ServiceName string `mapstructure:"service_name"` + ServiceVersion string `mapstructure:"service_version"` + Environment string `mapstructure:"environment"` +} + +type Shutdown func(context.Context) error + +func Setup(ctx context.Context, cfg Config) (Shutdown, error) { + otel.SetTextMapPropagator(propagation.TraceContext{}) + if !enabled(cfg) { + return func(context.Context) error { return nil }, nil + } + + opts := []otlptracegrpc.Option{} + if endpoint := strings.TrimSpace(cfg.Endpoint); endpoint != "" { + if strings.Contains(endpoint, "://") { + opts = append(opts, otlptracegrpc.WithEndpointURL(endpoint)) + } else { + opts = append(opts, otlptracegrpc.WithEndpoint(endpoint)) + } + } + if cfg.Insecure { + opts = append(opts, otlptracegrpc.WithInsecure()) + } + + exporter, err := otlptracegrpc.New(ctx, opts...) + if err != nil { + return nil, err + } + tp, ok := otel.GetTracerProvider().(*sdktrace.TracerProvider) + if !ok { + _ = exporter.Shutdown(ctx) + return nil, errors.New("全局 tracer provider 不是 SDK provider") + } + + processor := sdktrace.NewBatchSpanProcessor( + &resourceExporter{SpanExporter: exporter, resource: buildResource(cfg)}, + sdktrace.WithMaxQueueSize(2048), + sdktrace.WithMaxExportBatchSize(512), + sdktrace.WithBatchTimeout(5*time.Second), + sdktrace.WithExportTimeout(5*time.Second), + ) + tp.RegisterSpanProcessor(processor) + return processor.Shutdown, nil +} + +func enabled(cfg Config) bool { + if strings.EqualFold(strings.TrimSpace(os.Getenv("OTEL_SDK_DISABLED")), "true") || !cfg.Enabled { + return false + } + if strings.TrimSpace(cfg.Endpoint) != "" { + return true + } + return strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")) != "" || + strings.TrimSpace(os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")) != "" +} + +func buildResource(cfg Config) *resource.Resource { + attrs := []attribute.KeyValue{attribute.String("service.name", cfg.ServiceName)} + if cfg.ServiceVersion != "" { + attrs = append(attrs, attribute.String("service.version", cfg.ServiceVersion)) + } + if cfg.Environment != "" { + attrs = append(attrs, attribute.String("deployment.environment.name", cfg.Environment)) + } + return resource.NewSchemaless(attrs...) +} + +type resourceExporter struct { + sdktrace.SpanExporter + resource *resource.Resource +} + +func (e *resourceExporter) ExportSpans(ctx context.Context, spans []sdktrace.ReadOnlySpan) error { + wrapped := make([]sdktrace.ReadOnlySpan, len(spans)) + for i, span := range spans { + wrapped[i] = resourceSpan{ + ReadOnlySpan: span, + resource: e.resource, + attributes: sanitizeAttributes(span.Attributes()), + events: sanitizeEvents(span.Events()), + links: sanitizeLinks(span.Links()), + status: sanitizeStatus(span.Status()), + } + } + return e.SpanExporter.ExportSpans(ctx, wrapped) +} + +type resourceSpan struct { + sdktrace.ReadOnlySpan + resource *resource.Resource + attributes []attribute.KeyValue + events []sdktrace.Event + links []sdktrace.Link + status sdktrace.Status +} + +func (s resourceSpan) Resource() *resource.Resource { + return s.resource +} + +func (s resourceSpan) Attributes() []attribute.KeyValue { + return s.attributes +} + +func (s resourceSpan) Events() []sdktrace.Event { + return s.events +} + +func (s resourceSpan) Links() []sdktrace.Link { + return s.links +} + +func (s resourceSpan) Status() sdktrace.Status { + return s.status +} diff --git a/docs/architecture/adr/0001-backend-distributed-tracing.md b/docs/architecture/adr/0001-backend-distributed-tracing.md new file mode 100644 index 000000000..3aaff596a --- /dev/null +++ b/docs/architecture/adr/0001-backend-distributed-tracing.md @@ -0,0 +1,189 @@ +# ADR-0001：MonkeyCode Backend 分布式追踪 + +- 状态：已接受 +- 日期：2026-07-16 +- 决策者：MonkeyCode 团队 + +## 背景 + +MonkeyCode Backend 是用户任务进入服务端后的业务入口，并通过 HTTP、WebSocket 等方式调用 Taskflow。当前排障主要依赖业务日志以及 `task_id`、`session_id`、`request_id` 等字段，无法稳定还原一次操作在 MonkeyCode 与 Taskflow 之间的技术因果关系。 + +本次建设从 MonkeyCode Backend 开始，不改动 Web、Desktop、Mobile，也不改动 VM Agent 或 Coding Agent。客户端请求进入 MonkeyCode 后创建新的 Trace；Taskflow 到 Agent 的交互由 Taskflow 记录为黑盒边界。 + +## 决策 + +采用 OpenTelemetry 和 W3C Trace Context 建设 MonkeyCode Backend 到 Taskflow 的分布式追踪。一个 MonkeyCode 任务不是一条持续数小时的 Trace，而是由多条短生命周期 Trace 组成，并通过 `monkeycode.task.id` 聚合。同步调用使用父子 Span，异步阶段使用 Span Link。 + +```text +公网客户端 + │ 不信任外部 traceparent + ▼ +MonkeyCode Backend ── traceparent ──▶ Taskflow ──▶ Agent 黑盒边界 + ▲ │ + └──────── traceparent ─────────────┘ + 回调 + +MonkeyCode / Taskflow ── OTLP ──▶ OpenTelemetry Collector ──▶ Tempo +MonkeyCode / Taskflow ── structured logs ──▶ VictoriaLogs +``` + +### Trace 入口与信任边界 + +- MonkeyCode 的公网业务接口忽略客户端传入的 `traceparent` 和 `tracestate`，创建新的根 Trace。 +- 未来若由受控网关创建 Trace，只能信任经过网关清洗并重新注入的上下文。 +- Taskflow 内部回调端点允许提取经过认证的 Taskflow Trace Context。 +- 不使用 OpenTelemetry Baggage 传播业务字段。 + +### 传播规则 + +- MonkeyCode 调用 Taskflow 时，通过 W3C `traceparent` 和 `tracestate` 传播上下文。 +- HTTP 使用标准 Header；WebSocket 仅在握手阶段传播连接上下文。 +- Token、日志片段、终端字节流和心跳不逐条创建 Span。 +- `task_id`、`session_id`、`request_id`、`vm_id` 继续通过现有参数和消息体传递，由两端写入 Span 属性与结构化日志。 +- Taskflow 发起的异步回调形成新的 Trace，MonkeyCode 将回调处理作为该 Trace 的下游 Span。 + +### Span 粒度 + +只追踪跨边界调用和关键业务阶段，不为普通 Go 函数创建 Span。 + +MonkeyCode 的核心 Span 包括： + +- 接收任务创建、启动、停止、重启和模型切换请求； +- 用户、项目、模型与配额校验； +- 创建或更新任务记录； +- 调用 Taskflow； +- 建立 Task Live、Control、Terminal 等长连接； +- 接收 Taskflow 回调； +- 数据库、Redis 和外部服务调用。 + +健康检查、心跳、轮询和流式数据块默认不产生业务 Span。 + +### 属性规范 + +优先采用 OpenTelemetry Semantic Conventions。自定义属性使用以下名称： + +| 属性 | 含义 | +| --- | --- | +| `monkeycode.task.id` | MonkeyCode 任务 ID | +| `monkeycode.agent.session.id` | 一次 Agent 执行会话 ID | +| `monkeycode.request.id` | 一次业务命令或交互 ID | +| `monkeycode.project.id` | MonkeyCode 项目 ID | +| `taskflow.vm.id` | Taskflow 虚拟机 ID | +| `taskflow.terminal.session.id` | 终端会话 ID | +| `task.outcome` | `succeeded`、`failed`、`cancelled` 或 `rejected` | + +现有业务字段、协议字段和数据库字段保持不变，仅在遥测层统一命名。 + +### 数据安全 + +Trace 属性采用严格允许名单。允许采集服务名、版本、环境、路由模板、方法、状态码、规范化业务 ID、操作阶段、重试次数和清洗后的错误分类。 + +以下内容禁止进入 Trace： + +- Prompt、模型回复、代码、文件内容和完整文件路径； +- 请求体、响应体和 URL 查询参数； +- Git 仓库地址； +- Authorization、Cookie、Token 和密钥； +- 用户名、邮箱、手机号和客户端 IP； +- SQL 参数、Redis Key 和 Value； +- 可能包含第三方响应正文的原始错误信息。 + +数据库 Span 只保留操作类型、表名、耗时和错误分类；Redis Span 只保留命令名。 + +### 日志关联 + +服务运行日志继续写入 VictoriaLogs，不引入 Loki 作为运行日志后端。日志处理器从 `context.Context` 提取并补充： + +- `trace_id`； +- `span_id`； +- `monkeycode.task.id`； +- `monkeycode.agent.session.id`； +- `monkeycode.request.id`； +- `service.name`。 + +Grafana 必须支持从 VictoriaLogs 日志按 `trace_id` 打开 Tempo Trace，也支持从 Tempo 按 `trace_id` 和时间范围查询 VictoriaLogs。 + +Taskflow 中用于任务输出流的 Loki 业务能力不属于本次迁移范围。 + +### 错误语义 + +- 系统异常、网络失败、超时、数据库失败和 Taskflow 调度失败标记为 `Error`。 +- HTTP 5xx 与 gRPC `Internal`、`Unavailable`、`DeadlineExceeded` 标记为 `Error`。 +- 用户主动取消记录 `task.outcome=cancelled`，不标记系统错误。 +- 参数校验失败、未授权、配额不足和并发上限记录 `task.outcome=rejected` 及规范化原因，不标记系统错误。 +- 原始错误正文不进入 Trace。 + +### 导出与故障隔离 + +- 使用异步批量导出器，通过 OTLP 上报 OpenTelemetry Collector。 +- Collector Endpoint 未配置或追踪关闭时使用 Noop Provider。 +- 队列有界；队列满、网络阻塞或 Collector 不可用时丢弃 Trace，不阻塞业务。 +- 应用关闭时限时 Flush，超时后继续退出。 +- 导出失败、丢弃 Span 和队列使用率通过 Prometheus 暴露，并对错误日志限频。 + +优先使用 OpenTelemetry 标准环境变量配置导出地址、协议、服务名、环境和资源属性。应用配置只补充启停开关与无法由标准变量表达的限制。 + +### 采样与保留 + +应用将候选 Span 发送给 Collector，由 Collector 执行尾部采样： + +- 开发、测试环境保留 100%； +- 生产正常请求保留 10%； +- 错误、超时、关键任务操作和慢请求保留 100%； +- 健康检查、心跳和轮询默认丢弃； +- HTTP 慢请求初始阈值为 2 秒，其他操作使用独立阈值。 + +生产 Trace 保留 7 天，测试环境保留 3 天，本地开发保留 24 小时。VictoriaLogs 继续使用现有保留策略。 + +### 性能预算 + +- HTTP P95 额外延迟不超过 5 ms； +- 应用 CPU 增量不超过 3%； +- 单实例遥测队列内存上限 64 MiB； +- WebSocket 和 gRPC 长连接吞吐下降不超过 2%； +- 不增加请求路径中的同步遥测网络调用； +- Collector 故障时业务接口和长连接不得明显恶化。 + +### 上线顺序 + +1. 部署 Collector、Tempo 和 Grafana 数据源，配置 VictoriaLogs 双向跳转。 +2. Taskflow 上线接收端埋点和日志关联，默认关闭导出。 +3. MonkeyCode 上线服务端埋点与 Taskflow 上下文传播。 +4. 测试环境 100% 开启并完成验收。 +5. 生产单实例或小流量开启，观察性能和导出健康度。 +6. 满足性能预算后全量启用生产采样策略。 + +### 验收标准 + +- 正常创建任务可看到 MonkeyCode 接入、数据操作、Taskflow 调用和调度阶段； +- Taskflow 异步结果可通过 Span Link 和 `task_id` 关联到创建阶段； +- Taskflow 不可用、VM 创建失败和 Agent 超时正确保留并标记错误； +- 用户取消和业务拒绝不会误报为系统错误； +- WebSocket 重连形成独立连接 Span，并可通过业务 ID 聚合； +- VictoriaLogs 与 Tempo 可以双向跳转； +- Collector 停止时业务功能、延迟和长连接保持正常； +- Trace 中不存在禁止采集的数据； +- 性能测试满足既定预算。 + +## 影响 + +### 正面影响 + +- 可以从 `trace_id` 还原一次同步调用，也可以从 `task_id` 聚合长生命周期任务的多条 Trace。 +- 日志与 Trace 形成统一排障入口。 +- Agent 无需改动，降低协议和发布风险。 +- 遥测故障与业务故障隔离。 + +### 代价 + +- 需要维护 Collector、Tempo、采样规则和 Grafana 配置。 +- 异步关联需要 Taskflow 持久化起始 Trace Context。 +- 关键异步代码必须正确传递 `context.Context`，日志才能自动关联。 +- 尾部采样会增加 Collector 的入口流量与短期内存压力。 + +## 不在本次范围 + +- Web、Desktop、Mobile 客户端埋点； +- VM Agent、Coding Agent 内部追踪或协议修改； +- 业务任务日志从 Loki 迁移到 VictoriaLogs； +- 在应用仓库内维护生产级 Tempo 集群。 diff --git a/docs/architecture/observability-glossary.md b/docs/architecture/observability-glossary.md new file mode 100644 index 000000000..7f2035664 --- /dev/null +++ b/docs/architecture/observability-glossary.md @@ -0,0 +1,61 @@ +# MonkeyCode 可观测性术语表 + +## 领域标识 + +| 术语 | 遥测字段 | 定义 | +| --- | --- | --- | +| 任务 | `monkeycode.task.id` | MonkeyCode 中的长生命周期业务聚合。一个任务可关联多条 Trace | +| Agent 执行会话 | `monkeycode.agent.session.id` | 一次 Agent 执行上下文，不等同于终端会话 | +| 业务请求 | `monkeycode.request.id` | 一次命令、交互或请求响应配对标识，不等同于 Trace ID | +| 项目 | `monkeycode.project.id` | MonkeyCode 项目标识 | +| 虚拟机 | `taskflow.vm.id` | Taskflow 管理的执行环境标识 | +| 终端会话 | `taskflow.terminal.session.id` | 一次终端连接会话，不等同于 Agent 执行会话 | + +现有协议和数据库中的 `task_id`、`session_id`、`request_id` 不重命名。写入 Trace 时必须映射到语义明确的规范字段。 + +## 追踪术语 + +| 术语 | 定义 | +| --- | --- | +| Trace | 一次有边界的技术调用链。一个 MonkeyCode 任务通常包含多条 Trace | +| Span | Trace 中的一个操作，例如处理 HTTP 请求或调用 Taskflow | +| Trace Context | 由 `traceparent` 和 `tracestate` 表示的跨进程因果上下文 | +| Span Link | 将异步 Trace 与先前 Trace 建立关系，不形成长期父子 Span | +| Baggage | 可随 Trace Context 传播的业务数据容器。本设计不使用它传播业务 ID | +| 根 Trace | 没有上游父 Span 的 Trace。MonkeyCode 公网入口会创建新的根 Trace | +| 连接 Span | 描述 WebSocket 或 gRPC 长连接建立、断开与重连的 Span，不覆盖全部流消息生命周期 | +| 边界 Span | Taskflow 对 Agent 发送、等待或接收行为的观测。Agent 内部仍是黑盒 | +| 尾部采样 | Collector 收到完整候选 Trace 后，根据错误、耗时和属性决定是否保留 | + +## 存储与查询 + +| 组件 | 职责 | +| --- | --- | +| OpenTelemetry SDK | 在 MonkeyCode、Taskflow 进程内创建并批量导出 Span | +| OpenTelemetry Collector | 接收 OTLP、执行尾部采样并转发 Trace | +| Tempo | 存储 Trace 数据 | +| VictoriaLogs | 存储 MonkeyCode、Taskflow 的结构化运行日志 | +| Grafana | 查询 Trace 和日志，提供二者之间的跳转入口 | +| Loki 任务日志 | Taskflow 当前用于任务输出流的业务存储，不等同于服务运行日志 | + +## 结果语义 + +| 结果 | `task.outcome` | Span 状态 | +| --- | --- | --- | +| 成功 | `succeeded` | Unset | +| 系统或执行失败 | `failed` | Error | +| 用户主动取消 | `cancelled` | Unset | +| 参数、权限、配额或并发限制拒绝 | `rejected` | Unset | + +## 信任边界 + +- 公网客户端传入的 Trace Context 默认不可信，MonkeyCode 创建新的根 Trace。 +- 经过认证的 MonkeyCode 到 Taskflow 请求可以传播 Trace Context。 +- 经过认证的 Taskflow 回调可以向 MonkeyCode 传播 Trace Context。 +- Agent 不传播 Trace Context，本次不改动 Agent。 + +## 数据分类 + +允许进入 Trace 的数据包括服务资源信息、路由模板、状态码、规范化业务 ID、耗时、重试次数和错误分类。 + +Prompt、模型回复、代码、文件内容、完整路径、请求响应正文、查询参数、仓库地址、认证信息、个人信息、SQL 参数、Redis Key/Value 和原始第三方错误正文属于禁止采集数据。