diff --git a/ai/prompts/agentAct.txt b/ai/prompts/agentAct.txt index 35e326c61..b546e51ba 100644 --- a/ai/prompts/agentAct.txt +++ b/ai/prompts/agentAct.txt @@ -12,4 +12,16 @@ Tool selection policy Important: You must not answer anything in natural language. Always call one of the provided tools if applicable. -If you cannot find a suitable tool, return an empty tool call. \ No newline at end of file +If you cannot find a suitable tool, return an empty tool call. + +Observability tool policy +- Call `get_observability_capabilities` before the first metrics or trace query when provider availability is unknown. +- Use `query_prometheus` for a current snapshot and `query_prometheus_range` for trends. Range queries must stay within 24 hours and use a step of at least 15 seconds. +- Use metric and label names that exist in Dubbo Admin dashboards. Important labels include `application_name`, `instance`, `pod`, `interface`, `method`, and `version`. +- Common provider metrics include `dubbo_provider_qps_total`, `dubbo_provider_requests_total`, `dubbo_provider_requests_succeed_total`, `dubbo_provider_requests_failed_total`, `dubbo_provider_requests_timeout_total`, `dubbo_provider_requests_processing`, `dubbo_provider_rt_avg_milliseconds_aggregate`, `dubbo_provider_rt_milliseconds_p95`, and `dubbo_provider_rt_milliseconds_p99`. +- Equivalent consumer metrics use the `dubbo_consumer_` prefix. Thread-pool signals include `dubbo_thread_pool_active_size`, `dubbo_thread_pool_core_size`, and `dubbo_thread_pool_queue_size`. +- Counters must be evaluated with `rate` or `increase` over a bounded window. For success rate, divide successful requests by total requests and protect against an empty denominator. +- Begin with an application-level aggregation, then narrow to `instance` or `pod`; narrow to `interface` and `method` only when service-level evidence requires it. +- Do not repeat an identical query after a successful response. Change the dimension or time window only to test a stated hypothesis. +- Call `get_trace_by_id` only with a TraceID supplied by the user or returned by another tool. Use parentSpanId, serviceName, durationMicros, and error status to locate propagation and bottlenecks. +- Never invent metric names, label values, TraceIDs, provider names, or tool results. diff --git a/ai/prompts/agentObserve.txt b/ai/prompts/agentObserve.txt index e92351bd8..e5d959533 100644 --- a/ai/prompts/agentObserve.txt +++ b/ai/prompts/agentObserve.txt @@ -9,6 +9,11 @@ Given conversation history and tool outputs, return a short structured decision. - Set `heartbeat=false` when current evidence is enough to answer the user. - If the same or similar RAG/tool search failed multiple times, stop and answer with available knowledge. - Do not repeat long raw tool content. +- For metrics, cite the evaluated time or range, key labels, and values that support the conclusion. +- For traces, cite the relevant service, operation, span ID, parent span ID, duration, and error status. +- Distinguish explicitly between no data, provider/query failure, and evidence of a healthy state. +- A single metric spike is not a root cause. Continue when logs, trace evidence, or a comparison dimension is still needed to validate the hypothesis. +- Stop when the available metrics and trace evidence identify the affected scope and the most likely failure boundary, or when configured observability cannot provide further evidence. # Latest-Turn Rules (STRICT) - You must answer the latest user message in this session only. diff --git a/ai/prompts/agentThink.txt b/ai/prompts/agentThink.txt index 97bff743b..cca2ddfc5 100644 --- a/ai/prompts/agentThink.txt +++ b/ai/prompts/agentThink.txt @@ -34,6 +34,15 @@ Choose domain by question: - Dubbo topics -> Dubbo docs tools - General topics -> pick the most relevant domain first +# Observability Diagnosis Rules +- For live Dubbo incidents, prefer runtime evidence over documentation retrieval. +- If observability availability is unknown, suggest `get_observability_capabilities` first. +- For metric questions, use `query_prometheus` for the current state, then `query_prometheus_range` only when a trend or incident window is needed. +- When the user supplies a TraceID, or a log result exposes one, suggest `get_trace_by_id` before guessing where the failure occurred. +- Correlate evidence in this order when available: service metadata -> metrics -> logs -> trace spans. +- Treat missing data, tool failure, and healthy data as different outcomes. Missing data is not evidence of health. +- Prefer narrow application, instance, interface, and method filters. Never suggest an unbounded high-cardinality PromQL query. + # Output Constraint Return JSON only. No markdown, no code fence, no extra text. diff --git a/app/dubbo-admin/dubbo-admin.yaml b/app/dubbo-admin/dubbo-admin.yaml index b348772ce..4161c39a9 100644 --- a/app/dubbo-admin/dubbo-admin.yaml +++ b/app/dubbo-admin/dubbo-admin.yaml @@ -43,6 +43,16 @@ observability: # type: loki # endpoint: http://localhost:3100 # tenant: "" + # [Optional] trace providers used by MCP diagnosis tools. + # The bundled monitoring manifests use the Jaeger query API on port 16686. + # tracing: + # defaultProvider: jaeger-main + # providers: + # - name: jaeger-main + # type: jaeger + # endpoint: http://jaeger.monitoring.svc:16686 + # bearerToken: "" + # tenant: "" # [Optional] config for console. console: diff --git a/pkg/config/observability/config.go b/pkg/config/observability/config.go index 920e8a31c..b50470644 100644 --- a/pkg/config/observability/config.go +++ b/pkg/config/observability/config.go @@ -35,6 +35,8 @@ type Config struct { Prometheus string `json:"prometheus" yaml:"prometheus"` // Logs configures the log query provider. Logs *LogsConfig `json:"logs,omitempty" yaml:"logs,omitempty"` + // Tracing configures trace query providers used by diagnostic tools. + Tracing *TracingConfig `json:"tracing,omitempty" yaml:"tracing,omitempty"` GrafanaBaseURL *url.URL `json:"-" yaml:"-"` PrometheusBaseURL *url.URL `json:"-" yaml:"-"` @@ -62,9 +64,21 @@ func (c *Config) Validate() error { return err } } + if c.Tracing != nil { + if err := c.Tracing.Validate(); err != nil { + return err + } + } return nil } +func (c *Config) Sanitize() { + if c == nil || c.Tracing == nil { + return + } + c.Tracing.Sanitize() +} + func DefaultObservabilityConfig() *Config { return &Config{} } diff --git a/pkg/config/observability/tracing.go b/pkg/config/observability/tracing.go new file mode 100644 index 000000000..452154297 --- /dev/null +++ b/pkg/config/observability/tracing.go @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package observability + +import ( + "fmt" + "net/url" + + "github.com/duke-git/lancet/v2/strutil" + + "github.com/apache/dubbo-admin/pkg/common/bizerror" + "github.com/apache/dubbo-admin/pkg/config" +) + +// TraceProviderType identifies a supported trace query backend. +type TraceProviderType string + +const ( + // TraceProviderJaeger queries traces through the Jaeger query API. + TraceProviderJaeger TraceProviderType = "jaeger" +) + +// TracingConfig selects the default trace provider and contains provider definitions. +type TracingConfig struct { + DefaultProvider string `json:"defaultProvider" yaml:"defaultProvider"` + Providers []TraceProviderConfig `json:"providers" yaml:"providers"` +} + +// TraceProviderConfig contains connection and optional authentication settings for one backend. +type TraceProviderConfig struct { + Name string `json:"name" yaml:"name"` + Type TraceProviderType `json:"type" yaml:"type"` + Endpoint string `json:"endpoint" yaml:"endpoint"` + BearerToken string `json:"bearerToken,omitempty" yaml:"bearerToken,omitempty"` + Tenant string `json:"tenant,omitempty" yaml:"tenant,omitempty"` +} + +// Validate verifies provider uniqueness, endpoint safety, and default-provider selection. +func (c *TracingConfig) Validate() error { + if c == nil || len(c.Providers) == 0 { + return nil + } + if strutil.IsBlank(c.DefaultProvider) { + return bizerror.New(bizerror.ConfigError, "default trace provider is required") + } + + seen := make(map[string]struct{}, len(c.Providers)) + foundDefault := false + for _, provider := range c.Providers { + if strutil.IsBlank(provider.Name) { + return bizerror.New(bizerror.ConfigError, "trace provider name is required") + } + if _, ok := seen[provider.Name]; ok { + return bizerror.New(bizerror.ConfigError, fmt.Sprintf("duplicate trace provider name: %s", provider.Name)) + } + seen[provider.Name] = struct{}{} + if provider.Name == c.DefaultProvider { + foundDefault = true + } + if provider.Type != TraceProviderJaeger { + return bizerror.New(bizerror.ConfigError, fmt.Sprintf("unsupported trace provider type: %s", provider.Type)) + } + if err := validateTraceEndpoint(provider.Endpoint); err != nil { + return err + } + } + if !foundDefault { + return bizerror.New(bizerror.ConfigError, fmt.Sprintf("default trace provider %q is not configured", c.DefaultProvider)) + } + return nil +} + +func validateTraceEndpoint(endpoint string) error { + if strutil.IsBlank(endpoint) { + return bizerror.New(bizerror.ConfigError, "trace provider endpoint is required") + } + parsed, err := url.Parse(endpoint) + if err != nil { + return bizerror.Wrap(err, bizerror.ConfigError, fmt.Sprintf("invalid trace provider endpoint: %s", endpoint)) + } + if (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" { + return bizerror.New(bizerror.ConfigError, fmt.Sprintf("invalid trace provider endpoint: %s", endpoint)) + } + return nil +} + +// Default returns the configured default trace provider. +func (c *TracingConfig) Default() (TraceProviderConfig, bool) { + if c == nil { + return TraceProviderConfig{}, false + } + return c.Get(c.DefaultProvider) +} + +// Get returns a named trace provider. +func (c *TracingConfig) Get(name string) (TraceProviderConfig, bool) { + if c == nil { + return TraceProviderConfig{}, false + } + for _, provider := range c.Providers { + if provider.Name == name { + return provider, true + } + } + return TraceProviderConfig{}, false +} + +// Sanitize masks provider credentials before configuration is displayed. +func (c *TracingConfig) Sanitize() { + if c == nil { + return + } + for i := range c.Providers { + if c.Providers[i].BearerToken != "" { + c.Providers[i].BearerToken = config.SanitizedValue + } + } +} diff --git a/pkg/config/observability/tracing_test.go b/pkg/config/observability/tracing_test.go new file mode 100644 index 000000000..43eccb9ea --- /dev/null +++ b/pkg/config/observability/tracing_test.go @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package observability + +import ( + "testing" + + "github.com/apache/dubbo-admin/pkg/config" +) + +func TestTracingConfigValidate(t *testing.T) { + cfg := &TracingConfig{ + DefaultProvider: "jaeger-main", + Providers: []TraceProviderConfig{{ + Name: "jaeger-main", Type: TraceProviderJaeger, Endpoint: "http://jaeger:16686", + }}, + } + if err := cfg.Validate(); err != nil { + t.Fatalf("expected valid tracing config, got %v", err) + } + + cfg.Providers[0].Endpoint = "jaeger:16686" + if err := cfg.Validate(); err == nil { + t.Fatal("expected endpoint without scheme to be rejected") + } +} + +func TestTracingConfigRejectsDuplicateAndMissingDefault(t *testing.T) { + cfg := &TracingConfig{ + DefaultProvider: "missing", + Providers: []TraceProviderConfig{ + {Name: "same", Type: TraceProviderJaeger, Endpoint: "http://jaeger:16686"}, + {Name: "same", Type: TraceProviderJaeger, Endpoint: "http://jaeger-2:16686"}, + }, + } + if err := cfg.Validate(); err == nil { + t.Fatal("expected duplicate provider names to be rejected") + } + + cfg.Providers = cfg.Providers[:1] + if err := cfg.Validate(); err == nil { + t.Fatal("expected missing default provider to be rejected") + } +} + +func TestObservabilitySanitizeTraceToken(t *testing.T) { + cfg := &Config{Tracing: &TracingConfig{Providers: []TraceProviderConfig{{BearerToken: "secret"}}}} + cfg.Sanitize() + if got := cfg.Tracing.Providers[0].BearerToken; got != config.SanitizedValue { + t.Fatalf("expected sanitized token, got %q", got) + } +} + +func TestTracingConfigRejectsUnsupportedProvider(t *testing.T) { + cfg := &TracingConfig{ + DefaultProvider: "tempo-main", + Providers: []TraceProviderConfig{{ + Name: "tempo-main", Type: TraceProviderType("tempo"), Endpoint: "http://tempo:3200", + }}, + } + if err := cfg.Validate(); err == nil { + t.Fatal("expected unsupported provider to be rejected") + } +} diff --git a/pkg/mcp/DEVELOPMENT.md b/pkg/mcp/DEVELOPMENT.md index afe77ecdd..79f4121de 100644 --- a/pkg/mcp/DEVELOPMENT.md +++ b/pkg/mcp/DEVELOPMENT.md @@ -407,3 +407,32 @@ const ( - 使用 `common.JsonResult` 返回结果 - 使用 `common.ErrorResult` 处理错误 - 使用 `common.DefaultPageSize` 和 `common.DefaultPageNumber` 常量 + +## 可观测性诊断工具 + +Dubbo Admin MCP 提供以下只读诊断工具： + +| 工具 | 数据源 | 用途 | +|------|--------|------| +| `query_prometheus` | Prometheus | 执行即时 PromQL 查询 | +| `query_prometheus_range` | Prometheus | 执行最长 24 小时的区间 PromQL 查询 | +| `get_trace_by_id` | Jaeger | 按 TraceID 查询规范化调用链 | +| `get_observability_capabilities` | 本地配置 | 查询数据源和查询限制 | + +Prometheus 工具复用 `observability.prometheus`。区间查询的 `step` 不得低于 15 秒，返回结果最多包含 100 个时间序列和 5000 个采样点。结果被裁剪时，响应中的 `truncated` 为 `true`。 + +Jaeger 查询需要配置 trace provider： + +```yaml +observability: + tracing: + defaultProvider: jaeger-main + providers: + - name: jaeger-main + type: jaeger + endpoint: http://jaeger.monitoring.svc:16686 + bearerToken: "" # 可选 + tenant: "" # 可选，通过 X-Scope-OrgID 发送 +``` + +诊断工具不会记录完整 PromQL、Bearer Token 或完整 trace attributes。工具只读，不提供流量、配置或部署变更能力。 diff --git a/pkg/mcp/register.go b/pkg/mcp/register.go index 92ccbab36..2d8856a58 100644 --- a/pkg/mcp/register.go +++ b/pkg/mcp/register.go @@ -21,6 +21,8 @@ import ( "github.com/apache/dubbo-admin/pkg/mcp/common" "github.com/apache/dubbo-admin/pkg/mcp/tools" logtools "github.com/apache/dubbo-admin/pkg/mcp/tools/log" + metrictools "github.com/apache/dubbo-admin/pkg/mcp/tools/metrics" + tracetools "github.com/apache/dubbo-admin/pkg/mcp/tools/trace" ) // RegisterTools 注册所有 MCP 工具 @@ -370,4 +372,44 @@ func RegisterTools(server *Server) { }, Handler: logtools.GetLogCapabilities, }) + + server.RegisterTool(&common.ToolDef{ + Name: "query_prometheus", + Description: "Execute an instant PromQL query and return normalized metric series", + InputSchema: common.InputSchema{ + Type: "object", + Properties: metrictools.InstantProperties(), + Required: []string{"query"}, + }, + Handler: metrictools.QueryPrometheus, + }) + + server.RegisterTool(&common.ToolDef{ + Name: "query_prometheus_range", + Description: "Execute a bounded PromQL range query for trend and incident-window analysis", + InputSchema: common.InputSchema{ + Type: "object", + Properties: metrictools.RangeProperties(), + Required: []string{"query", "startTime", "endTime", "step"}, + }, + Handler: metrictools.QueryPrometheusRange, + }) + + server.RegisterTool(&common.ToolDef{ + Name: "get_trace_by_id", + Description: "Query a distributed trace by TraceID and return services, span relationships, latency, and error status", + InputSchema: common.InputSchema{ + Type: "object", + Properties: tracetools.Properties(), + Required: []string{"traceId"}, + }, + Handler: tracetools.GetTraceByID, + }) + + server.RegisterTool(&common.ToolDef{ + Name: "get_observability_capabilities", + Description: "Report Prometheus and trace providers, capabilities, and enforced query limits", + InputSchema: common.InputSchema{Type: "object"}, + Handler: tools.GetObservabilityCapabilities, + }) } diff --git a/pkg/mcp/register_test.go b/pkg/mcp/register_test.go index c9806f737..5d423b657 100644 --- a/pkg/mcp/register_test.go +++ b/pkg/mcp/register_test.go @@ -55,3 +55,40 @@ func TestRegisterServiceDetailTools(t *testing.T) { } } } + +func TestRegisterObservabilityTools(t *testing.T) { + server := NewServer("test", "dev") + RegisterTools(server) + + tests := []struct { + name string + required []string + }{ + {name: "query_prometheus", required: []string{"query"}}, + {name: "query_prometheus_range", required: []string{"query", "startTime", "endTime", "step"}}, + {name: "get_trace_by_id", required: []string{"traceId"}}, + {name: "get_observability_capabilities"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + tool, ok := server.tools[test.name] + if !ok { + t.Fatalf("tool %q is not registered", test.name) + } + if tool.Handler == nil { + t.Fatalf("tool %q has no handler", test.name) + } + if len(tool.InputSchema.Required) != len(test.required) { + t.Fatalf("tool %q required fields = %v, want %v", test.name, tool.InputSchema.Required, test.required) + } + for i, field := range test.required { + if tool.InputSchema.Required[i] != field { + t.Fatalf("tool %q required fields = %v, want %v", test.name, tool.InputSchema.Required, test.required) + } + if _, ok := tool.InputSchema.Properties[field]; !ok { + t.Fatalf("tool %q is missing schema property %q", test.name, field) + } + } + }) + } +} diff --git a/pkg/mcp/server.go b/pkg/mcp/server.go index e1719afb2..ba5b588ca 100644 --- a/pkg/mcp/server.go +++ b/pkg/mcp/server.go @@ -24,8 +24,10 @@ import ( "io" "net/http" "sync" + "time" consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/core/logger" "github.com/gin-gonic/gin" "github.com/apache/dubbo-admin/pkg/mcp/common" @@ -43,9 +45,9 @@ type Server struct { // NewServer 创建 MCP 服务器 func NewServer(name, version string) *Server { return &Server{ - name: name, + name: name, version: version, - tools: make(map[string]*common.ToolDef), + tools: make(map[string]*common.ToolDef), } } @@ -217,8 +219,10 @@ func (s *Server) handleToolsCall(req *common.JSONRPCRequest) *common.JSONRPCResp return s.newErrorResponse(req.ID, common.ErrCodeInvalidParams, err.Error()) } + startedAt := time.Now() result, err := tool.Handler(ctx, arguments) if err != nil { + logger.Sugar().Warnw("MCP tool call failed", "tool", name, "request_id", req.ID, "elapsed_ms", time.Since(startedAt).Milliseconds(), "error", err) return &common.JSONRPCResponse{ JSONRPC: common.JSONRPCVersion, ID: req.ID, @@ -228,6 +232,15 @@ func (s *Server) handleToolsCall(req *common.JSONRPCRequest) *common.JSONRPCResp }, } } + if result == nil { + logger.Sugar().Errorw("MCP tool returned nil result", "tool", name, "request_id", req.ID, "elapsed_ms", time.Since(startedAt).Milliseconds()) + return s.newErrorResponse(req.ID, common.ErrCodeInternalError, "Tool returned no result") + } + if result.IsError { + logger.Sugar().Warnw("MCP tool call completed with error", "tool", name, "request_id", req.ID, "elapsed_ms", time.Since(startedAt).Milliseconds(), "content_blocks", len(result.Content)) + } else { + logger.Sugar().Infow("MCP tool call completed", "tool", name, "request_id", req.ID, "elapsed_ms", time.Since(startedAt).Milliseconds(), "content_blocks", len(result.Content)) + } return &common.JSONRPCResponse{ JSONRPC: common.JSONRPCVersion, diff --git a/pkg/mcp/tools/metrics/model.go b/pkg/mcp/tools/metrics/model.go new file mode 100644 index 000000000..d2a33553a --- /dev/null +++ b/pkg/mcp/tools/metrics/model.go @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package metrics + +// Sample is one Prometheus timestamp-value pair. Value remains a string to preserve NaN and infinities. +type Sample struct { + Timestamp float64 `json:"timestamp"` + Value string `json:"value"` +} + +// Series contains labels and either an instant value or range values. +type Series struct { + Labels map[string]string `json:"labels"` + Value *Sample `json:"value,omitempty"` + Values []Sample `json:"values,omitempty"` +} + +// QueryResponse is the stable, bounded MCP representation of a Prometheus query result. +type QueryResponse struct { + QueryType string `json:"queryType"` + ResultType string `json:"resultType"` + Series []Series `json:"series,omitempty"` + Value *Sample `json:"value,omitempty"` + Warnings []string `json:"warnings,omitempty"` + Truncated bool `json:"truncated"` + ReturnedSeries int `json:"returnedSeries"` + ReturnedSamples int `json:"returnedSamples"` +} + +// QueryLimits describes server-enforced PromQL resource limits. +type QueryLimits struct { + TimeoutSeconds int `json:"timeoutSeconds"` + MaxQueryBytes int `json:"maxQueryBytes"` + MaxRange string `json:"maxRange"` + MinStep string `json:"minStep"` + MaxSeries int `json:"maxSeries"` + MaxSamples int `json:"maxSamples"` + MaxResponseBytes int64 `json:"maxResponseBytes"` +} diff --git a/pkg/mcp/tools/metrics/prometheus.go b/pkg/mcp/tools/metrics/prometheus.go new file mode 100644 index 000000000..294b6f9c2 --- /dev/null +++ b/pkg/mcp/tools/metrics/prometheus.go @@ -0,0 +1,329 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package metrics + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "math" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "github.com/apache/dubbo-admin/pkg/core/logger" +) + +const ( + queryTimeout = 15 * time.Second + maxQueryBytes = 8 * 1024 + maxQueryRange = 24 * time.Hour + minQueryStep = 15 * time.Second + maxSeries = 100 + maxSamples = 5000 + maxResponseBytes = int64(4 * 1024 * 1024) + maxErrorBodyBytes = int64(4096) +) + +type prometheusClient struct { + baseURL *url.URL + client *http.Client +} + +var prometheusTransport = &http.Transport{ + Proxy: http.ProxyFromEnvironment, + MaxIdleConns: 20, + MaxIdleConnsPerHost: 10, + IdleConnTimeout: 90 * time.Second, + ResponseHeaderTimeout: queryTimeout, +} + +type prometheusResponse struct { + Status string `json:"status"` + Data struct { + ResultType string `json:"resultType"` + Result json.RawMessage `json:"result"` + } `json:"data"` + ErrorType string `json:"errorType,omitempty"` + Error string `json:"error,omitempty"` + Warnings []string `json:"warnings,omitempty"` +} + +type prometheusSeries struct { + Metric map[string]string `json:"metric"` + Value []json.RawMessage `json:"value,omitempty"` + Values [][]json.RawMessage `json:"values,omitempty"` +} + +func newPrometheusClient(baseURL *url.URL) *prometheusClient { + return &prometheusClient{ + baseURL: baseURL, + client: &http.Client{ + Timeout: queryTimeout, + Transport: prometheusTransport, + }, + } +} + +func (c *prometheusClient) instant(ctx context.Context, query, evaluationTime string) (*QueryResponse, error) { + values := url.Values{"query": []string{query}} + if evaluationTime != "" { + if _, err := parsePrometheusTime("time", evaluationTime); err != nil { + return nil, err + } + values.Set("time", evaluationTime) + } + return c.query(ctx, "instant", "/api/v1/query", values) +} + +func (c *prometheusClient) rangeQuery(ctx context.Context, query, startRaw, endRaw, stepRaw string) (*QueryResponse, error) { + start, err := parsePrometheusTime("startTime", startRaw) + if err != nil { + return nil, err + } + end, err := parsePrometheusTime("endTime", endRaw) + if err != nil { + return nil, err + } + if !end.After(start) { + return nil, fmt.Errorf("endTime must be after startTime") + } + if end.Sub(start) > maxQueryRange { + return nil, fmt.Errorf("query range must not exceed %s", maxQueryRange) + } + step, err := parsePrometheusStep(stepRaw) + if err != nil { + return nil, err + } + if step < minQueryStep { + return nil, fmt.Errorf("step must be at least %s", minQueryStep) + } + + values := url.Values{ + "query": []string{query}, + "start": []string{startRaw}, + "end": []string{endRaw}, + "step": []string{stepRaw}, + } + return c.query(ctx, "range", "/api/v1/query_range", values) +} + +func (c *prometheusClient) query(ctx context.Context, queryType, path string, values url.Values) (*QueryResponse, error) { + query := values.Get("query") + if strings.TrimSpace(query) == "" { + return nil, fmt.Errorf("query is required") + } + if len(query) > maxQueryBytes { + return nil, fmt.Errorf("query must not exceed %d bytes", maxQueryBytes) + } + + endpoint := *c.baseURL + endpoint.Path = strings.TrimRight(endpoint.Path, "/") + path + startedAt := time.Now() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint.String(), strings.NewReader(values.Encode())) + if err != nil { + return nil, fmt.Errorf("failed to create prometheus request: %w", err) + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.Header.Set("Accept", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + logPrometheusFailure(queryType, query, startedAt, 0, "transport", err) + return nil, fmt.Errorf("prometheus query failed: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes)) + logPrometheusFailure(queryType, query, startedAt, resp.StatusCode, "upstream_status", nil) + return nil, fmt.Errorf("prometheus query failed with status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + + body, err := readBoundedBody(resp.Body, maxResponseBytes) + if err != nil { + logPrometheusFailure(queryType, query, startedAt, resp.StatusCode, "response_limit", err) + return nil, fmt.Errorf("failed to read prometheus response: %w", err) + } + var upstream prometheusResponse + if err := json.Unmarshal(body, &upstream); err != nil { + logPrometheusFailure(queryType, query, startedAt, resp.StatusCode, "decode", err) + return nil, fmt.Errorf("failed to decode prometheus response: %w", err) + } + if upstream.Status != "success" { + message := upstream.Error + if message == "" { + message = "unknown prometheus error" + } + errorType := upstream.ErrorType + if errorType == "" { + errorType = "unknown" + } + logPrometheusFailure(queryType, query, startedAt, resp.StatusCode, errorType, nil) + return nil, fmt.Errorf("prometheus %s error: %s", errorType, message) + } + result, err := normalizePrometheusResult(queryType, &upstream) + if err != nil { + logPrometheusFailure(queryType, query, startedAt, resp.StatusCode, "normalize", err) + return nil, err + } + logger.Sugar().Infow("prometheus query completed", "query_type", queryType, "query_hash", queryHash(query), "elapsed_ms", time.Since(startedAt).Milliseconds(), "result_type", result.ResultType, "series", result.ReturnedSeries, "samples", result.ReturnedSamples, "truncated", result.Truncated) + return result, nil +} + +func normalizePrometheusResult(queryType string, upstream *prometheusResponse) (*QueryResponse, error) { + result := &QueryResponse{ + QueryType: queryType, + ResultType: upstream.Data.ResultType, + Warnings: upstream.Warnings, + Series: []Series{}, + } + switch upstream.Data.ResultType { + case "vector", "matrix": + var rawSeries []prometheusSeries + if err := json.Unmarshal(upstream.Data.Result, &rawSeries); err != nil { + return nil, fmt.Errorf("failed to decode prometheus %s result: %w", upstream.Data.ResultType, err) + } + for i, raw := range rawSeries { + if i >= maxSeries || result.ReturnedSamples >= maxSamples { + result.Truncated = true + break + } + series := Series{Labels: raw.Metric} + if upstream.Data.ResultType == "vector" { + sample, err := decodeSample(raw.Value) + if err != nil { + return nil, err + } + series.Value = &sample + result.ReturnedSamples++ + } else { + for _, rawSample := range raw.Values { + if result.ReturnedSamples >= maxSamples { + result.Truncated = true + break + } + sample, err := decodeSample(rawSample) + if err != nil { + return nil, err + } + series.Values = append(series.Values, sample) + result.ReturnedSamples++ + } + } + result.Series = append(result.Series, series) + result.ReturnedSeries++ + } + case "scalar", "string": + var raw []json.RawMessage + if err := json.Unmarshal(upstream.Data.Result, &raw); err != nil { + return nil, fmt.Errorf("failed to decode prometheus %s result: %w", upstream.Data.ResultType, err) + } + sample, err := decodeSample(raw) + if err != nil { + return nil, err + } + result.Value = &sample + result.ReturnedSamples = 1 + default: + return nil, fmt.Errorf("unsupported prometheus result type %q", upstream.Data.ResultType) + } + return result, nil +} + +func decodeSample(raw []json.RawMessage) (Sample, error) { + if len(raw) != 2 { + return Sample{}, fmt.Errorf("invalid prometheus sample: expected timestamp and value") + } + var timestamp float64 + if err := json.Unmarshal(raw[0], ×tamp); err != nil { + return Sample{}, fmt.Errorf("invalid prometheus sample timestamp: %w", err) + } + var value string + if err := json.Unmarshal(raw[1], &value); err != nil { + return Sample{}, fmt.Errorf("invalid prometheus sample value: %w", err) + } + return Sample{Timestamp: timestamp, Value: value}, nil +} + +func parsePrometheusTime(field, value string) (time.Time, error) { + if parsed, err := time.Parse(time.RFC3339Nano, value); err == nil { + return parsed, nil + } + seconds, err := strconv.ParseFloat(value, 64) + if err != nil || math.IsNaN(seconds) || math.IsInf(seconds, 0) || seconds > float64(math.MaxInt64) || seconds < float64(math.MinInt64) { + return time.Time{}, fmt.Errorf("%s must be RFC3339 or Unix seconds", field) + } + whole, fraction := math.Modf(seconds) + return time.Unix(int64(whole), int64(fraction*float64(time.Second))), nil +} + +func parsePrometheusStep(value string) (time.Duration, error) { + if parsed, err := time.ParseDuration(value); err == nil { + return parsed, nil + } + seconds, err := strconv.ParseFloat(value, 64) + if err != nil || seconds <= 0 { + return 0, fmt.Errorf("step must be a positive duration or number of seconds") + } + return time.Duration(seconds * float64(time.Second)), nil +} + +func readBoundedBody(reader io.Reader, limit int64) ([]byte, error) { + body, err := io.ReadAll(io.LimitReader(reader, limit+1)) + if err != nil { + return nil, err + } + if int64(len(body)) > limit { + return nil, fmt.Errorf("response exceeds %d bytes", limit) + } + return body, nil +} + +func queryHash(query string) string { + digest := sha256.Sum256([]byte(query)) + return hex.EncodeToString(digest[:6]) +} + +func logPrometheusFailure(queryType, query string, startedAt time.Time, status int, errorType string, err error) { + fields := []any{"query_type", queryType, "query_hash", queryHash(query), "elapsed_ms", time.Since(startedAt).Milliseconds(), "error_type", errorType} + if status != 0 { + fields = append(fields, "upstream_status", status) + } + if err != nil { + fields = append(fields, "error", err) + } + logger.Sugar().Warnw("prometheus query failed", fields...) +} + +// Limits returns the resource limits enforced by the Prometheus MCP tools. +func Limits() QueryLimits { + return QueryLimits{ + TimeoutSeconds: int(queryTimeout / time.Second), + MaxQueryBytes: maxQueryBytes, + MaxRange: maxQueryRange.String(), + MinStep: minQueryStep.String(), + MaxSeries: maxSeries, + MaxSamples: maxSamples, + MaxResponseBytes: maxResponseBytes, + } +} diff --git a/pkg/mcp/tools/metrics/prometheus_test.go b/pkg/mcp/tools/metrics/prometheus_test.go new file mode 100644 index 000000000..73b973bb4 --- /dev/null +++ b/pkg/mcp/tools/metrics/prometheus_test.go @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package metrics + +import ( + "context" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" +) + +func TestPrometheusInstantQueryNormalizesVector(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/api/v1/query" { + t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path) + } + if err := r.ParseForm(); err != nil { + t.Fatal(err) + } + if r.Form.Get("query") != "up" { + t.Fatalf("unexpected query %q", r.Form.Get("query")) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"status":"success","data":{"resultType":"vector","result":[{"metric":{"job":"dubbo"},"value":[1710000000.5,"NaN"]}]},"warnings":["partial data"]}`)) + })) + defer server.Close() + + baseURL, _ := url.Parse(server.URL) + result, err := newPrometheusClient(baseURL).instant(context.Background(), "up", "") + if err != nil { + t.Fatal(err) + } + if result.ResultType != "vector" || result.ReturnedSeries != 1 || result.ReturnedSamples != 1 { + t.Fatalf("unexpected result: %#v", result) + } + if result.Series[0].Value.Value != "NaN" || result.Series[0].Labels["job"] != "dubbo" { + t.Fatalf("unexpected normalized series: %#v", result.Series[0]) + } +} + +func TestPrometheusRangeValidation(t *testing.T) { + baseURL, _ := url.Parse("http://prometheus:9090") + client := newPrometheusClient(baseURL) + _, err := client.rangeQuery(context.Background(), "rate(x[5m])", "2026-01-01T00:00:00Z", "2026-01-02T00:00:01Z", "15s") + if err == nil || !strings.Contains(err.Error(), "must not exceed") { + t.Fatalf("expected range limit error, got %v", err) + } + _, err = client.rangeQuery(context.Background(), "rate(x[5m])", "2026-01-01T00:00:00Z", "2026-01-01T01:00:00Z", "5s") + if err == nil || !strings.Contains(err.Error(), "at least") { + t.Fatalf("expected step limit error, got %v", err) + } +} + +func TestNormalizePrometheusMatrixTruncatesSamples(t *testing.T) { + values := make([]string, 0, maxSamples+1) + for i := 0; i < maxSamples+1; i++ { + values = append(values, `[1710000000,"1"]`) + } + upstream := &prometheusResponse{Status: "success"} + upstream.Data.ResultType = "matrix" + upstream.Data.Result = []byte(`[{"metric":{"job":"dubbo"},"values":[` + strings.Join(values, ",") + `]}]`) + result, err := normalizePrometheusResult("range", upstream) + if err != nil { + t.Fatal(err) + } + if !result.Truncated || result.ReturnedSamples != maxSamples { + t.Fatalf("expected sample truncation, got %#v", result) + } +} + +func TestPrometheusReturnsUpstreamError(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusUnprocessableEntity) + _, _ = w.Write([]byte("bad query")) + })) + defer server.Close() + baseURL, _ := url.Parse(server.URL) + _, err := newPrometheusClient(baseURL).instant(context.Background(), "bad(", "") + if err == nil || !strings.Contains(err.Error(), "status 422") { + t.Fatalf("expected upstream status error, got %v", err) + } +} + +func TestReadBoundedBodyRejectsOversizedResponse(t *testing.T) { + _, err := readBoundedBody(strings.NewReader("12345"), 4) + if err == nil || !strings.Contains(err.Error(), "exceeds") { + t.Fatalf("expected response limit error, got %v", err) + } +} diff --git a/pkg/mcp/tools/metrics/tools.go b/pkg/mcp/tools/metrics/tools.go new file mode 100644 index 000000000..8af6ba953 --- /dev/null +++ b/pkg/mcp/tools/metrics/tools.go @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package metrics + +import ( + "context" + "fmt" + + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/mcp/common" +) + +// QueryPrometheus executes an instant PromQL query against the configured Prometheus endpoint. +func QueryPrometheus(ctx consolectx.Context, args map[string]any) (*common.ToolResult, error) { + client, requestCtx, err := clientFromContext(ctx) + if err != nil { + return common.ErrorResult(err), nil + } + helper := common.NewArgsHelper(args) + resp, err := client.instant(requestCtx, helper.GetString("query", ""), helper.GetString("time", "")) + if err != nil { + return common.ErrorResult(err), nil + } + return common.JsonResult(resp) +} + +// QueryPrometheusRange executes a bounded PromQL range query. +func QueryPrometheusRange(ctx consolectx.Context, args map[string]any) (*common.ToolResult, error) { + client, requestCtx, err := clientFromContext(ctx) + if err != nil { + return common.ErrorResult(err), nil + } + helper := common.NewArgsHelper(args) + resp, err := client.rangeQuery( + requestCtx, + helper.GetString("query", ""), + helper.GetString("startTime", ""), + helper.GetString("endTime", ""), + helper.GetString("step", ""), + ) + if err != nil { + return common.ErrorResult(err), nil + } + return common.JsonResult(resp) +} + +func clientFromContext(ctx consolectx.Context) (*prometheusClient, context.Context, error) { + if ctx == nil || ctx.Config().Observability == nil || ctx.Config().Observability.PrometheusBaseURL == nil { + return nil, nil, fmt.Errorf("prometheus is not configured") + } + requestCtx := context.Background() + if ctx.AppContext() != nil { + requestCtx = ctx.AppContext() + } + return newPrometheusClient(ctx.Config().Observability.PrometheusBaseURL), requestCtx, nil +} + +// InstantProperties returns the MCP input schema properties for instant queries. +func InstantProperties() map[string]common.PropertyDef { + return map[string]common.PropertyDef{ + "query": {Type: "string", Description: "PromQL expression"}, + "time": {Type: "string", Description: "Optional evaluation time in RFC3339 or Unix seconds"}, + } +} + +// RangeProperties returns the MCP input schema properties for range queries. +func RangeProperties() map[string]common.PropertyDef { + return map[string]common.PropertyDef{ + "query": {Type: "string", Description: "PromQL expression"}, + "startTime": {Type: "string", Description: "Start time in RFC3339 or Unix seconds"}, + "endTime": {Type: "string", Description: "End time in RFC3339 or Unix seconds"}, + "step": {Type: "string", Description: "Query step, for example 30s or 60"}, + } +} diff --git a/pkg/mcp/tools/observability.go b/pkg/mcp/tools/observability.go new file mode 100644 index 000000000..2914a08c3 --- /dev/null +++ b/pkg/mcp/tools/observability.go @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package tools + +import ( + "sort" + + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/mcp/common" + metrictools "github.com/apache/dubbo-admin/pkg/mcp/tools/metrics" +) + +type observabilityCapabilities struct { + PrometheusConfigured bool `json:"prometheusConfigured"` + TraceConfigured bool `json:"traceConfigured"` + TraceProviders []traceCapability `json:"traceProviders"` + Limits metrictools.QueryLimits `json:"limits"` +} + +type traceCapability struct { + Name string `json:"name"` + Type string `json:"type"` +} + +// GetObservabilityCapabilities reports configured providers and enforced PromQL limits without secrets. +func GetObservabilityCapabilities(ctx consolectx.Context, _ map[string]any) (*common.ToolResult, error) { + resp := observabilityCapabilities{Limits: metrictools.Limits()} + if ctx == nil || ctx.Config().Observability == nil { + return common.JsonResult(resp) + } + cfg := ctx.Config().Observability + resp.PrometheusConfigured = cfg.PrometheusBaseURL != nil + resp.TraceConfigured = cfg.Tracing != nil && len(cfg.Tracing.Providers) > 0 + if cfg.Tracing != nil { + for _, provider := range cfg.Tracing.Providers { + resp.TraceProviders = append(resp.TraceProviders, traceCapability{Name: provider.Name, Type: string(provider.Type)}) + } + sort.Slice(resp.TraceProviders, func(i, j int) bool { return resp.TraceProviders[i].Name < resp.TraceProviders[j].Name }) + } + return common.JsonResult(resp) +} diff --git a/pkg/mcp/tools/trace/jaeger.go b/pkg/mcp/tools/trace/jaeger.go new file mode 100644 index 000000000..354c5ebf2 --- /dev/null +++ b/pkg/mcp/tools/trace/jaeger.go @@ -0,0 +1,406 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package trace + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "sort" + "strconv" + "strings" + "time" + "unicode/utf8" + + observabilitycfg "github.com/apache/dubbo-admin/pkg/config/observability" + "github.com/apache/dubbo-admin/pkg/core/logger" +) + +const ( + traceQueryTimeout = 15 * time.Second + maxTraceResponseBytes = int64(8 * 1024 * 1024) + maxTraceErrorBodyBytes = int64(4096) + maxTraceSpans = 500 + maxTraceRoots = 50 + maxTraceServices = 100 + maxSpanAttributes = 32 + maxSpanEvents = 10 + maxEventAttributes = 16 + maxAttributeValueBytes = 512 + tenantHeader = "X-Scope-OrgID" +) + +type jaegerProvider struct { + config observabilitycfg.TraceProviderConfig + baseURL *url.URL + client *http.Client +} + +var jaegerTransport = &http.Transport{ + Proxy: http.ProxyFromEnvironment, + MaxIdleConns: 20, + MaxIdleConnsPerHost: 10, + IdleConnTimeout: 90 * time.Second, + ResponseHeaderTimeout: traceQueryTimeout, +} + +type jaegerResponse struct { + Data []jaegerTrace `json:"data"` + Errors []struct { + Code int `json:"code"` + Message string `json:"msg"` + } `json:"errors,omitempty"` +} + +type jaegerTrace struct { + TraceID string `json:"traceID"` + Spans []jaegerSpan `json:"spans"` + Processes map[string]jaegerProcess `json:"processes"` +} + +type jaegerSpan struct { + TraceID string `json:"traceID"` + SpanID string `json:"spanID"` + OperationName string `json:"operationName"` + References []jaegerReference `json:"references"` + StartTime int64 `json:"startTime"` + Duration int64 `json:"duration"` + Tags []jaegerKeyValue `json:"tags"` + Logs []jaegerLog `json:"logs"` + ProcessID string `json:"processID"` + Process *jaegerProcess `json:"process,omitempty"` +} + +type jaegerReference struct { + RefType string `json:"refType"` + TraceID string `json:"traceID"` + SpanID string `json:"spanID"` +} + +type jaegerProcess struct { + ServiceName string `json:"serviceName"` + Tags []jaegerKeyValue `json:"tags"` +} + +type jaegerLog struct { + Timestamp int64 `json:"timestamp"` + Fields []jaegerKeyValue `json:"fields"` +} + +type jaegerKeyValue struct { + Key string `json:"key"` + Type string `json:"type"` + Value any `json:"value"` +} + +func newJaegerProvider(cfg observabilitycfg.TraceProviderConfig) (*jaegerProvider, error) { + baseURL, err := url.Parse(cfg.Endpoint) + if err != nil { + return nil, fmt.Errorf("invalid jaeger endpoint: %w", err) + } + return &jaegerProvider{ + config: cfg, + baseURL: baseURL, + client: &http.Client{ + Timeout: traceQueryTimeout, + Transport: jaegerTransport, + }, + }, nil +} + +func (p *jaegerProvider) GetTraceByID(ctx context.Context, traceID string) (*Trace, error) { + endpoint := *p.baseURL + endpoint.Path = strings.TrimRight(endpoint.Path, "/") + "/api/traces/" + traceID + startedAt := time.Now() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint.String(), nil) + if err != nil { + return nil, fmt.Errorf("failed to create jaeger request: %w", err) + } + req.Header.Set("Accept", "application/json") + if p.config.BearerToken != "" { + req.Header.Set("Authorization", "Bearer "+p.config.BearerToken) + } + if p.config.Tenant != "" { + req.Header.Set(tenantHeader, p.config.Tenant) + } + + resp, err := p.client.Do(req) + if err != nil { + p.logFailure(traceID, startedAt, 0, "transport", err) + return nil, fmt.Errorf("jaeger trace query failed: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound { + p.logFailure(traceID, startedAt, resp.StatusCode, "not_found", nil) + return nil, fmt.Errorf("trace %s was not found", traceID) + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + body, _ := io.ReadAll(io.LimitReader(resp.Body, maxTraceErrorBodyBytes)) + p.logFailure(traceID, startedAt, resp.StatusCode, "upstream_status", nil) + return nil, fmt.Errorf("jaeger trace query failed with status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + body, err := readTraceBody(resp.Body) + if err != nil { + p.logFailure(traceID, startedAt, resp.StatusCode, "response_limit", err) + return nil, err + } + var upstream jaegerResponse + if err := json.Unmarshal(body, &upstream); err != nil { + p.logFailure(traceID, startedAt, resp.StatusCode, "decode", err) + return nil, fmt.Errorf("failed to decode jaeger response: %w", err) + } + if len(upstream.Errors) > 0 { + p.logFailure(traceID, startedAt, resp.StatusCode, "upstream_error", nil) + return nil, fmt.Errorf("jaeger trace query failed: %s", upstream.Errors[0].Message) + } + if len(upstream.Data) == 0 { + p.logFailure(traceID, startedAt, resp.StatusCode, "not_found", nil) + return nil, fmt.Errorf("trace %s was not found", traceID) + } + if len(upstream.Data) != 1 || !strings.EqualFold(upstream.Data[0].TraceID, traceID) { + p.logFailure(traceID, startedAt, resp.StatusCode, "inconsistent_response", nil) + return nil, fmt.Errorf("jaeger returned an inconsistent response for trace %s", traceID) + } + result := normalizeJaegerTrace(upstream.Data[0]) + logger.Sugar().Infow("trace query completed", "provider", p.config.Name, "engine", p.config.Type, "trace_id", maskedTraceID(traceID), "elapsed_ms", time.Since(startedAt).Milliseconds(), "spans", result.SpanCount, "services", result.ServiceCount, "error_spans", result.ErrorSpanCount, "truncated", result.Truncated) + return result, nil +} + +func normalizeJaegerTrace(input jaegerTrace) *Trace { + result := &Trace{TraceID: input.TraceID, SourceEngine: "jaeger", SpanCount: len(input.Spans), Spans: []Span{}} + serviceSet := map[string]struct{}{} + var minStart, maxEnd int64 + spans := input.Spans + sort.SliceStable(spans, func(i, j int) bool { return spans[i].StartTime < spans[j].StartTime }) + for _, rawSpan := range spans { + process := input.Processes[rawSpan.ProcessID] + if rawSpan.Process != nil { + process = *rawSpan.Process + } + if process.ServiceName != "" { + serviceSet[process.ServiceName] = struct{}{} + } + if errorSpan, _ := spanStatus(rawSpan.Tags); errorSpan { + result.ErrorSpanCount++ + } + if parentSpanID(rawSpan.References) == "" { + if len(result.RootSpanIDs) < maxTraceRoots { + result.RootSpanIDs = append(result.RootSpanIDs, rawSpan.SpanID) + } else { + result.Truncated = true + } + } + end := rawSpan.StartTime + rawSpan.Duration + if minStart == 0 || rawSpan.StartTime < minStart { + minStart = rawSpan.StartTime + } + if end > maxEnd { + maxEnd = end + } + } + if len(spans) > maxTraceSpans { + spans = spans[:maxTraceSpans] + result.Truncated = true + } + for _, rawSpan := range spans { + process := input.Processes[rawSpan.ProcessID] + if rawSpan.Process != nil { + process = *rawSpan.Process + } + span := normalizeJaegerSpan(rawSpan, process) + result.Spans = append(result.Spans, span) + if span.Truncated { + result.Truncated = true + } + } + result.ReturnedSpanCount = len(result.Spans) + for service := range serviceSet { + result.Services = append(result.Services, service) + } + sort.Strings(result.Services) + result.ServiceCount = len(result.Services) + if len(result.Services) > maxTraceServices { + result.Services = result.Services[:maxTraceServices] + result.Truncated = true + } + if minStart > 0 { + result.StartTime = jaegerMicrosTime(minStart) + result.DurationMicros = maxEnd - minStart + } + return result +} + +func normalizeJaegerSpan(input jaegerSpan, process jaegerProcess) Span { + attributes, attributesTruncated := normalizeKeyValues(input.Tags, maxSpanAttributes) + for _, processTag := range process.Tags { + key := "process." + processTag.Key + if _, exists := attributes[key]; exists { + continue + } + if len(attributes) >= maxSpanAttributes { + attributesTruncated = true + break + } + value := formatAttributeValue(processTag.Value) + if truncatedValue, truncated := truncateUTF8(value, maxAttributeValueBytes); truncated { + value = truncatedValue + attributesTruncated = true + } + attributes[key] = value + } + span := Span{ + SpanID: input.SpanID, + ParentSpanID: parentSpanID(input.References), + ServiceName: process.ServiceName, + OperationName: input.OperationName, + StartTime: jaegerMicrosTime(input.StartTime), + DurationMicros: input.Duration, + Status: "UNSET", + Attributes: attributes, + Truncated: attributesTruncated, + } + span.Error, span.Status = spanStatus(input.Tags) + for i, logEntry := range input.Logs { + if i >= maxSpanEvents { + span.Truncated = true + break + } + eventAttributes, truncated := normalizeKeyValues(logEntry.Fields, maxEventAttributes) + span.Events = append(span.Events, Event{Timestamp: jaegerMicrosTime(logEntry.Timestamp), Attributes: eventAttributes}) + span.Truncated = span.Truncated || truncated + } + return span +} + +func normalizeKeyValues(values []jaegerKeyValue, limit int) (map[string]string, bool) { + result := make(map[string]string, min(limit, len(values))) + truncated := false + for _, item := range values { + if len(result) >= limit { + truncated = true + break + } + value := formatAttributeValue(item.Value) + if truncatedValue, valueTruncated := truncateUTF8(value, maxAttributeValueBytes); valueTruncated { + value = truncatedValue + truncated = true + } + result[item.Key] = value + } + return result, truncated +} + +func spanStatus(tags []jaegerKeyValue) (bool, string) { + status := "UNSET" + errorSpan := false + for _, tag := range tags { + key := strings.ToLower(tag.Key) + value := strings.ToLower(formatAttributeValue(tag.Value)) + switch key { + case "error": + if value == "true" || value == "1" { + errorSpan = true + status = "ERROR" + } + case "otel.status_code", "status.code": + if value == "error" { + errorSpan = true + status = "ERROR" + } else if value == "ok" && !errorSpan { + status = "OK" + } + } + } + return errorSpan, status +} + +func parentSpanID(references []jaegerReference) string { + for _, reference := range references { + if strings.EqualFold(reference.RefType, "CHILD_OF") { + return reference.SpanID + } + } + if len(references) > 0 { + return references[0].SpanID + } + return "" +} + +func formatAttributeValue(value any) string { + switch typed := value.(type) { + case string: + return typed + case bool: + return strconv.FormatBool(typed) + case float64: + return strconv.FormatFloat(typed, 'g', -1, 64) + default: + encoded, err := json.Marshal(typed) + if err != nil { + return fmt.Sprint(typed) + } + return string(encoded) + } +} + +func jaegerMicrosTime(value int64) string { + return time.Unix(0, value*int64(time.Microsecond)).UTC().Format(time.RFC3339Nano) +} + +func readTraceBody(reader io.Reader) ([]byte, error) { + body, err := io.ReadAll(io.LimitReader(reader, maxTraceResponseBytes+1)) + if err != nil { + return nil, fmt.Errorf("failed to read jaeger response: %w", err) + } + if int64(len(body)) > maxTraceResponseBytes { + return nil, fmt.Errorf("jaeger response exceeds %d bytes", maxTraceResponseBytes) + } + return body, nil +} + +func maskedTraceID(traceID string) string { + if len(traceID) <= 8 { + return traceID + } + return "..." + traceID[len(traceID)-8:] +} + +func truncateUTF8(value string, limit int) (string, bool) { + if len(value) <= limit { + return value, false + } + value = value[:limit] + for !utf8.ValidString(value) { + value = value[:len(value)-1] + } + return value, true +} + +func (p *jaegerProvider) logFailure(traceID string, startedAt time.Time, status int, errorType string, err error) { + fields := []any{"provider", p.config.Name, "engine", p.config.Type, "trace_id", maskedTraceID(traceID), "elapsed_ms", time.Since(startedAt).Milliseconds(), "error_type", errorType} + if status != 0 { + fields = append(fields, "upstream_status", status) + } + if err != nil { + fields = append(fields, "error", err) + } + logger.Sugar().Warnw("trace query failed", fields...) +} diff --git a/pkg/mcp/tools/trace/jaeger_test.go b/pkg/mcp/tools/trace/jaeger_test.go new file mode 100644 index 000000000..8132b611b --- /dev/null +++ b/pkg/mcp/tools/trace/jaeger_test.go @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package trace + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + observabilitycfg "github.com/apache/dubbo-admin/pkg/config/observability" +) + +const testTraceID = "faba6a688ea3070b1613f50fb081c578" + +func TestJaegerProviderQueriesAndNormalizesTrace(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/traces/"+testTraceID { + t.Fatalf("unexpected path %s", r.URL.Path) + } + if r.Header.Get("Authorization") != "Bearer secret" || r.Header.Get(tenantHeader) != "tenant-a" { + t.Fatalf("missing auth headers: %#v", r.Header) + } + _, _ = w.Write([]byte(`{"data":[{"traceID":"` + testTraceID + `","spans":[{"traceID":"` + testTraceID + `","spanID":"01","operationName":"root","references":[],"startTime":1000000,"duration":5000,"tags":[],"logs":[],"processID":"p1"},{"traceID":"` + testTraceID + `","spanID":"02","operationName":"child","references":[{"refType":"CHILD_OF","traceID":"` + testTraceID + `","spanID":"01"}],"startTime":1001000,"duration":2000,"tags":[{"key":"error","type":"bool","value":true}],"logs":[],"processID":"p2"}],"processes":{"p1":{"serviceName":"frontend","tags":[]},"p2":{"serviceName":"backend","tags":[]}}}]}`)) + })) + defer server.Close() + + provider, err := newJaegerProvider(observabilitycfg.TraceProviderConfig{ + Name: "jaeger-main", Type: observabilitycfg.TraceProviderJaeger, Endpoint: server.URL, BearerToken: "secret", Tenant: "tenant-a", + }) + if err != nil { + t.Fatal(err) + } + result, err := provider.GetTraceByID(context.Background(), testTraceID) + if err != nil { + t.Fatal(err) + } + if result.SpanCount != 2 || result.ReturnedSpanCount != 2 || result.ErrorSpanCount != 1 || result.ServiceCount != 2 { + t.Fatalf("unexpected trace summary: %#v", result) + } + if result.Spans[1].ParentSpanID != "01" || result.Spans[1].Status != "ERROR" { + t.Fatalf("unexpected child span: %#v", result.Spans[1]) + } +} + +func TestTraceIDValidation(t *testing.T) { + for _, value := range []string{"abc", "../trace", "1234567890abcdefg"} { + if traceIDPattern.MatchString(value) { + t.Fatalf("expected invalid trace ID %q", value) + } + } + for _, value := range []string{"1234567890abcdef", testTraceID} { + if !traceIDPattern.MatchString(value) { + t.Fatalf("expected valid trace ID %q", value) + } + } +} + +func TestJaegerProviderReturnsNotFound(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNotFound) + })) + defer server.Close() + provider, err := newJaegerProvider(observabilitycfg.TraceProviderConfig{ + Name: "jaeger-main", Type: observabilitycfg.TraceProviderJaeger, Endpoint: server.URL, + }) + if err != nil { + t.Fatal(err) + } + if _, err := provider.GetTraceByID(context.Background(), testTraceID); err == nil { + t.Fatal("expected trace not found error") + } +} + +func TestNormalizeJaegerTraceTruncatesSpans(t *testing.T) { + input := jaegerTrace{TraceID: testTraceID, Spans: make([]jaegerSpan, maxTraceSpans+1)} + for i := range input.Spans { + input.Spans[i] = jaegerSpan{SpanID: string(rune(i + 1)), StartTime: int64(i + 1), Duration: 1} + } + result := normalizeJaegerTrace(input) + if !result.Truncated || result.SpanCount != maxTraceSpans+1 || result.ReturnedSpanCount != maxTraceSpans { + t.Fatalf("expected trace truncation, got %#v", result) + } +} diff --git a/pkg/mcp/tools/trace/model.go b/pkg/mcp/tools/trace/model.go new file mode 100644 index 000000000..78d07ce60 --- /dev/null +++ b/pkg/mcp/tools/trace/model.go @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package trace + +// Trace is a backend-independent, bounded representation of a distributed trace. +type Trace struct { + TraceID string `json:"traceId"` + StartTime string `json:"startTime"` + DurationMicros int64 `json:"durationMicros"` + SpanCount int `json:"spanCount"` + ReturnedSpanCount int `json:"returnedSpanCount"` + ServiceCount int `json:"serviceCount"` + ErrorSpanCount int `json:"errorSpanCount"` + Services []string `json:"services"` + RootSpanIDs []string `json:"rootSpanIds"` + Spans []Span `json:"spans"` + SourceEngine string `json:"sourceEngine"` + Truncated bool `json:"truncated"` +} + +// Span contains the diagnostic fields needed to reconstruct latency and error propagation. +type Span struct { + SpanID string `json:"spanId"` + ParentSpanID string `json:"parentSpanId,omitempty"` + ServiceName string `json:"serviceName"` + OperationName string `json:"operationName"` + StartTime string `json:"startTime"` + DurationMicros int64 `json:"durationMicros"` + Status string `json:"status"` + Error bool `json:"error"` + Attributes map[string]string `json:"attributes,omitempty"` + Events []Event `json:"events,omitempty"` + Truncated bool `json:"truncated"` +} + +// Event represents a bounded Jaeger span log entry. +type Event struct { + Timestamp string `json:"timestamp"` + Attributes map[string]string `json:"attributes,omitempty"` +} diff --git a/pkg/mcp/tools/trace/provider.go b/pkg/mcp/tools/trace/provider.go new file mode 100644 index 000000000..863256af2 --- /dev/null +++ b/pkg/mcp/tools/trace/provider.go @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package trace + +import ( + "context" + "fmt" + + observabilitycfg "github.com/apache/dubbo-admin/pkg/config/observability" +) + +// Provider hides backend wire formats behind a stable trace lookup contract. +type Provider interface { + GetTraceByID(ctx context.Context, traceID string) (*Trace, error) +} + +func newProvider(cfg observabilitycfg.TraceProviderConfig) (Provider, error) { + switch cfg.Type { + case observabilitycfg.TraceProviderJaeger: + return newJaegerProvider(cfg) + default: + return nil, fmt.Errorf("unsupported trace provider type: %s", cfg.Type) + } +} diff --git a/pkg/mcp/tools/trace/tools.go b/pkg/mcp/tools/trace/tools.go new file mode 100644 index 000000000..d59318759 --- /dev/null +++ b/pkg/mcp/tools/trace/tools.go @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package trace + +import ( + "context" + "fmt" + "regexp" + + observabilitycfg "github.com/apache/dubbo-admin/pkg/config/observability" + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/mcp/common" +) + +var traceIDPattern = regexp.MustCompile(`^(?:[0-9a-fA-F]{16}|[0-9a-fA-F]{32})$`) + +// GetTraceByID queries a configured provider using a validated Jaeger trace identifier. +func GetTraceByID(ctx consolectx.Context, args map[string]any) (*common.ToolResult, error) { + helper := common.NewArgsHelper(args) + traceID := helper.GetString("traceId", "") + if !traceIDPattern.MatchString(traceID) { + return common.ErrorResult(fmt.Errorf("traceId must be a 16 or 32 character hexadecimal value")), nil + } + providerConfig, err := providerConfigFromContext(ctx, helper.GetString("provider", "")) + if err != nil { + return common.ErrorResult(err), nil + } + provider, err := newProvider(providerConfig) + if err != nil { + return common.ErrorResult(err), nil + } + requestCtx := context.Background() + if ctx.AppContext() != nil { + requestCtx = ctx.AppContext() + } + result, err := provider.GetTraceByID(requestCtx, traceID) + if err != nil { + return common.ErrorResult(err), nil + } + return common.JsonResult(result) +} + +func providerConfigFromContext(ctx consolectx.Context, name string) (observabilitycfg.TraceProviderConfig, error) { + if ctx == nil || ctx.Config().Observability == nil || ctx.Config().Observability.Tracing == nil { + return observabilitycfg.TraceProviderConfig{}, fmt.Errorf("trace provider is not configured") + } + tracing := ctx.Config().Observability.Tracing + if name == "" { + provider, ok := tracing.Default() + if !ok { + return observabilitycfg.TraceProviderConfig{}, fmt.Errorf("default trace provider is not configured") + } + return provider, nil + } + provider, ok := tracing.Get(name) + if !ok { + return observabilitycfg.TraceProviderConfig{}, fmt.Errorf("trace provider %q is not configured", name) + } + return provider, nil +} + +// Properties returns the MCP input schema properties for trace lookup. +func Properties() map[string]common.PropertyDef { + return map[string]common.PropertyDef{ + "traceId": {Type: "string", Description: "Jaeger 64-bit or 128-bit hexadecimal TraceID"}, + "provider": {Type: "string", Description: "Optional trace provider name; defaults to defaultProvider"}, + } +}