Copilot commented on code in PR #1468: URL: https://github.com/apache/dubbo-admin/pull/1468#discussion_r3214554727
########## pkg/mcp/tools/resource_search.go: ########## @@ -0,0 +1,291 @@ +/* + * 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 ( + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/console/model" + "github.com/apache/dubbo-admin/pkg/console/service" + "github.com/apache/dubbo-admin/pkg/mcp/types" + "github.com/apache/dubbo-admin/pkg/mcp/registry" +) + +// ResourceSearchRegistrar 搜索工具注册器 +type ResourceSearchRegistrar struct{} + +// searchExecutor 搜索执行器接口 +type searchExecutor interface { + execute(ctx consolectx.Context, keyword, mesh string, pageNumber, pageSize int) (*model.SearchPaginationResult, error) + buildResult(pagedResult *model.SearchPaginationResult, keyword string, pageSize, pageNumber int) map[string]any +} + +// RegisterTools 实现 ToolRegistrar 接口 +func (r *ResourceSearchRegistrar) RegisterTools(reg *registry.Registry) { + reg.Register(types.ToolDef{ + Name: "global_search", + Description: "全局搜索，支持搜索服务、实例、应用等资源。不传 keyword 返回所有数据", + InputSchema: types.InputSchema{ + Type: "object", + Required: []string{}, // keyword 改为可选，空值返回所有数据 + Properties: map[string]types.PropertyDef{ + "keyword": { + Type: "string", + Description: "搜索关键字，为空时返回所有数据", + }, + "searchType": { + Type: "string", + Description: "搜索类型: ip(按IP搜索实例), instanceName(按实例名搜索), appName(按应用名搜索), serviceName(按服务名搜索)", + Default: string(SearchTypeName), + Enum: []string{string(SearchTypeIP), string(SearchTypeInstanceName), string(SearchTypeAppName), string(SearchTypeName)}, + }, + "mesh": { + Type: "string", + Description: "Mesh 名称，默认使用配置中的默认 mesh", + }, + "pageSize": { + Type: "integer", + Description: "每页数量", + Default: DefaultPageSize, + }, + "pageNumber": { + Type: "integer", + Description: "页码，从 1 开始", + Default: DefaultPageNumber, + }, + }, + }, + Handler: GlobalSearch, + }) +} + +// GlobalSearch 全局搜索 +func GlobalSearch(ctx consolectx.Context, args map[string]any) (*types.ToolResult, error) { + helper := NewArgsHelper(args) + keyword := helper.GetString("keyword", "") + + searchType := SearchType(helper.GetString("searchType", string(SearchTypeName))) + mesh := GetMeshArg(ctx, args) + pageSize := helper.GetInt("pageSize", DefaultPageSize) + pageNumber := helper.GetInt("pageNumber", DefaultPageNumber) + + executor := getSearchExecutor(searchType) + result, err := executor.execute(ctx, keyword, mesh, pageNumber, pageSize) + if err != nil { + return ErrorResult(err), nil + } + + searchResult := executor.buildResult(result, keyword, pageSize, pageNumber) + searchResult["searchType"] = string(searchType) + + return JsonResult(searchResult) +} + +// getSearchExecutor 根据搜索类型获取对应的执行器 +func getSearchExecutor(searchType SearchType) searchExecutor { + executors := map[SearchType]searchExecutor{ + SearchTypeIP: &ipSearchExecutor{}, + SearchTypeInstanceName: &instanceNameSearchExecutor{}, + SearchTypeAppName: &appNameSearchExecutor{}, + SearchTypeName: &serviceNameSearchExecutor{}, + } + + if executor, ok := executors[searchType]; ok { + return executor + } + return &serviceNameSearchExecutor{} +} + +// ipSearchExecutor IP 搜索执行器 +type ipSearchExecutor struct{} + +func (e *ipSearchExecutor) execute(ctx consolectx.Context, keyword, mesh string, pageNumber, pageSize int) (*model.SearchPaginationResult, error) { + req := buildSearchReq(keyword, mesh, pageNumber, pageSize) + return service.SearchInstanceByIp(ctx, req) +} + +func (e *ipSearchExecutor) buildResult(pagedResult *model.SearchPaginationResult, keyword string, pageSize, pageNumber int) map[string]any { + instances, totalCount := extractInstances(pagedResult) + return map[string]any{ + "keyword": keyword, + "pageSize": pageSize, + "pageNumber": pageNumber, + "instances": instances, + "totalCount": totalCount, + } +} + +// instanceNameSearchExecutor 实例名搜索执行器 +type instanceNameSearchExecutor struct{} + +func (e *instanceNameSearchExecutor) execute(ctx consolectx.Context, keyword, mesh string, pageNumber, pageSize int) (*model.SearchPaginationResult, error) { + req := buildSearchReq(keyword, mesh, pageNumber, pageSize) + return service.SearchInstanceByName(ctx, req) +} + +func (e *instanceNameSearchExecutor) buildResult(pagedResult *model.SearchPaginationResult, keyword string, pageSize, pageNumber int) map[string]any { + instances, totalCount := extractInstances(pagedResult) + return map[string]any{ + "keyword": keyword, + "pageSize": pageSize, + "pageNumber": pageNumber, + "instances": instances, + "totalCount": totalCount, + } +} + +// appNameSearchExecutor 应用名搜索执行器 +type appNameSearchExecutor struct{} + +func (e *appNameSearchExecutor) execute(ctx consolectx.Context, keyword, mesh string, pageNumber, pageSize int) (*model.SearchPaginationResult, error) { + // 使用 SearchApplications 而不是 SearchApplicationsByKeywords + // SearchApplications 会正确处理空 keyword 的情况（返回所有应用的分页列表） + req := &model.ApplicationSearchReq{ + Keywords: keyword, + Mesh: mesh, + PageReq: BuildPageReq(pageNumber, pageSize), + } + return service.SearchApplications(ctx, req) +} + +func (e *appNameSearchExecutor) buildResult(pagedResult *model.SearchPaginationResult, keyword string, pageSize, pageNumber int) map[string]any { + apps := extractApplicationsFromResult(pagedResult) + return map[string]any{ + "keyword": keyword, + "pageSize": pageSize, + "pageNumber": pageNumber, + "applications": apps, + "totalCount": len(apps), Review Comment: `appNameSearchExecutor.buildResult` 将 `totalCount` 设为 `len(apps)`，只表示当前页数量而不是分页总数。建议在 `pagedResult.PageInfo != nil` 时返回 `pagedResult.PageInfo.Total`，避免调用方误判总量。 ########## pkg/config/app/admin.go: ########## @@ -18,82 +18,176 @@ package app import ( - "github.com/pkg/errors" + "github.com/duke-git/lancet/v2/slice" "go.uber.org/multierr" + "github.com/apache/dubbo-admin/pkg/common/bizerror" "github.com/apache/dubbo-admin/pkg/config" "github.com/apache/dubbo-admin/pkg/config/console" "github.com/apache/dubbo-admin/pkg/config/diagnostics" "github.com/apache/dubbo-admin/pkg/config/discovery" "github.com/apache/dubbo-admin/pkg/config/engine" - "github.com/apache/dubbo-admin/pkg/config/mode" + "github.com/apache/dubbo-admin/pkg/config/log" + "github.com/apache/dubbo-admin/pkg/config/observability" "github.com/apache/dubbo-admin/pkg/config/store" ) type AdminConfig struct { config.BaseConfig - // Mode in which dubbo admin is running. Available values are: "test", "global", "zone" - Mode mode.Mode `json:"mode" envconfig:"DUBBO_MODE"` + // Log configuration + Log *log.Config `json:"log" yaml:"log"` // Diagnostics configuration Review Comment: `AdminConfig` 不再包含运行模式字段（例如 `Mode`），但代码库中仍存在对 `cfg.Mode` 的引用（runtime builder / run 命令等），并且配置文件示例也包含 `mode:`。这会直接导致编译失败/配置不兼容。建议恢复该字段（及默认值/校验），或同步迁移所有引用点与配置。 ########## pkg/mcp/tools/service_discovery.go: ########## @@ -0,0 +1,243 @@ +/* + * 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 ( + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/console/model" + "github.com/apache/dubbo-admin/pkg/console/service" + "github.com/apache/dubbo-admin/pkg/mcp/types" + "github.com/apache/dubbo-admin/pkg/mcp/registry" +) + +// ServiceRegistrar 服务工具注册器 +type ServiceRegistrar struct{} + +// RegisterTools 实现 ToolRegistrar 接口 +func (r *ServiceRegistrar) RegisterTools(reg *registry.Registry) { + reg.Register(types.ToolDef{ + Name: "search_services", + Description: "搜索 Dubbo 服务，支持按服务名过滤和分页", + InputSchema: types.InputSchema{ + Type: "object", + Properties: map[string]types.PropertyDef{ + "keywords": { + Type: "string", + Description: "服务名搜索关键字，支持模糊匹配", + }, + "mesh": { + Type: "string", + Description: "Mesh 名称，默认使用配置中的默认 mesh", + }, + "pageSize": { + Type: "integer", + Description: "每页数量", + Default: DefaultPageSize, + }, + "pageNumber": { + Type: "integer", + Description: "页码，从 1 开始", + Default: DefaultPageNumber, + }, + }, + }, + Handler: SearchServices, + }) + + reg.Register(types.ToolDef{ + Name: "get_service_detail", + Description: "获取服务详情，包括服务分布和实例信息", + InputSchema: types.InputSchema{ + Type: "object", + Required: []string{"serviceName"}, + Properties: map[string]types.PropertyDef{ + "serviceName": { + Type: "string", + Description: "服务名称", + }, + "group": { + Type: "string", + Description: "服务分组", + Default: "", + }, + "version": { + Type: "string", + Description: "服务版本", + Default: "", + }, + "side": { + Type: "string", + Description: "服务端或消费者 (provider/consumer)", + Default: string(ServiceSideProvider), + Enum: []string{string(ServiceSideProvider), string(ServiceSideConsumer)}, + }, + "mesh": { + Type: "string", + Description: "Mesh 名称，默认使用配置中的默认 mesh", + }, + }, + }, + Handler: GetServiceDetail, + }) +} + +// SearchServices 搜索服务 +func SearchServices(ctx consolectx.Context, args map[string]any) (*types.ToolResult, error) { + helper := NewArgsHelper(args) + keywords := helper.GetString("keywords", "") + mesh := GetMeshArg(ctx, args) + pageSize := helper.GetInt("pageSize", DefaultPageSize) + pageNumber := helper.GetInt("pageNumber", DefaultPageNumber) + + req := &model.ServiceSearchReq{ + Keywords: keywords, + Mesh: mesh, + PageReq: BuildPageReq(pageNumber, pageSize), + } + + result, err := service.SearchServices(ctx, req) + if err != nil { + return ErrorResult(err), nil + } + + return buildServiceSearchResult(result, keywords, mesh, pageSize, pageNumber) +} + +// GetServiceDetail 获取服务详情 +func GetServiceDetail(ctx consolectx.Context, args map[string]any) (*types.ToolResult, error) { + helper := NewArgsHelper(args) + serviceName := helper.GetString("serviceName", "") Review Comment: `GetServiceDetail` 的 schema 声明 `serviceName` 为必需参数，但 handler 内部用 `GetString` 读取且未校验。绕过 JSON-RPC 校验直接调用 handler 时会用空 serviceName 发请求并产生不明确错误。建议在 handler 内使用 `GetRequiredString` 并返回清晰错误。 ########## pkg/mcp/transport/http/sse.go: ########## @@ -0,0 +1,191 @@ +/* + * 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 http + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync" + + "github.com/apache/dubbo-admin/pkg/mcp/core" +) + +// SSETransport Server-Sent Events传输层 +type SSETransport struct { + server *core.Server + clients map[*SSEClient]bool + mu sync.RWMutex + broadcast chan []byte +} + +// SSEClient SSE客户端连接 +type SSEClient struct { + id string + ch chan []byte + ctx context.Context + done context.CancelFunc +} + +// NewSSEClient 创建SSE客户端 +func NewSSEClient(id string) *SSEClient { + ctx, cancel := context.WithCancel(context.Background()) + return &SSEClient{ + id: id, + ch: make(chan []byte, 256), + ctx: ctx, + done: cancel, + } +} + +// NewSSETransport 创建SSE传输层 +func NewSSETransport(server *core.Server) *SSETransport { + return &SSETransport{ + server: server, + clients: make(map[*SSEClient]bool), + broadcast: make(chan []byte, 256), + } +} + +// HandleSSE 处理SSE连接 +func (t *SSETransport) HandleSSE(w http.ResponseWriter, r *http.Request) { + // 设置SSE headers + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("Access-Control-Allow-Origin", "*") + + // 创建客户端 + client := NewSSEClient(r.RemoteAddr) + + t.mu.Lock() + t.clients[client] = true + t.mu.Unlock() + + // 发送连接成功消息 + t.sendToClient(client, t.sseEvent("connected", "SSE connection established")) + + // 等待断开连接 + <-client.ctx.Done() + Review Comment: `HandleSSE` 只把事件写进 `client.ch`，随后阻塞等待 `<-client.ctx.Done()`；但函数内没有把 `client.ch` 的数据写到 `ResponseWriter`，也没有绑定 `r.Context()` 来在客户端断开时退出。这会导致 SSE 连接/测试挂起且客户端收不到事件。 ########## pkg/mcp/transport/http/http.go: ########## @@ -0,0 +1,168 @@ +/* + * 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 http + +import ( + "context" + "fmt" + "net/http" + "sync" + "time" + + "github.com/apache/dubbo-admin/pkg/mcp/core" +) + +// Transport HTTP传输层 +type Transport struct { + server *core.Server + httpServer *http.Server + handler *Handler + mu sync.RWMutex + started bool +} + +// Config HTTP传输配置 +type Config struct { + Host string + Port int + ReadTimeout time.Duration + WriteTimeout time.Duration + ShutdownTimeout time.Duration +} + +// DefaultConfig 返回默认配置 +func DefaultConfig() *Config { + return &Config{ + Host: "0.0.0.0", + Port: 8080, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + ShutdownTimeout: 10 * time.Second, + } +} + +// NewTransport 创建HTTP传输层 +func NewTransport(server *core.Server) *Transport { + return NewTransportWithConfig(server, DefaultConfig()) +} + +// NewTransportWithConfig 使用指定配置创建HTTP传输层 +func NewTransportWithConfig(server *core.Server, cfg *Config) *Transport { + handler := NewHandler(server) + + addr := fmt.Sprintf("%s:%d", cfg.Host, cfg.Port) + httpServer := &http.Server{ + Addr: addr, + Handler: handler, + ReadTimeout: cfg.ReadTimeout, + WriteTimeout: cfg.WriteTimeout, + } + + return &Transport{ + server: server, + httpServer: httpServer, + handler: handler, + } +} + +// Start 启动HTTP服务器（阻塞运行） +func (t *Transport) Start(ctx context.Context) error { + t.mu.Lock() + if t.started { + t.mu.Unlock() + return fmt.Errorf("transport already started") + } + t.started = true + t.mu.Unlock() + + // 启动服务器 + errCh := make(chan error, 1) + go func() { + if err := t.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + errCh <- err + } + }() + + // 等待上下文取消或错误 + select { + case <-ctx.Done(): + return t.Shutdown() + case err := <-errCh: + return err + } +} + +// StartAsync 异步启动HTTP服务器 +func (t *Transport) StartAsync(ctx context.Context) error { + t.mu.Lock() + if t.started { + t.mu.Unlock() + return fmt.Errorf("transport already started") + } + t.started = true + t.mu.Unlock() + + go func() { + if err := t.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + // 记录错误但不退出 + fmt.Printf("HTTP server error: %v\n", err) + } + }() + + return nil +} + +// Shutdown 关闭HTTP服务器 +func (t *Transport) Shutdown() error { + t.mu.Lock() + defer t.mu.Unlock() + + if !t.started { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + if err := t.httpServer.Shutdown(ctx); err != nil { + return fmt.Errorf("shutdown failed: %w", err) Review Comment: `Shutdown()` 使用硬编码 `10*time.Second`，即使用户设置了 `Config.ShutdownTimeout` 也不会生效。建议改为使用保存下来的 shutdown timeout 配置值（为 0 时回退默认值）。 ########## pkg/mcp/transport/http/http.go: ########## @@ -0,0 +1,168 @@ +/* + * 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 http + +import ( + "context" + "fmt" + "net/http" + "sync" + "time" + + "github.com/apache/dubbo-admin/pkg/mcp/core" +) + +// Transport HTTP传输层 +type Transport struct { + server *core.Server + httpServer *http.Server + handler *Handler + mu sync.RWMutex + started bool +} + +// Config HTTP传输配置 +type Config struct { + Host string + Port int + ReadTimeout time.Duration + WriteTimeout time.Duration + ShutdownTimeout time.Duration +} Review Comment: `Config` 里包含 `ShutdownTimeout`，但 `Transport` 没有保存该配置且后续也未使用，导致该配置项无效。建议把 shutdown timeout 存到 `Transport` 上并在 `Shutdown()` 使用。 ########## pkg/config/app/admin.go: ########## @@ -18,82 +18,176 @@ package app import ( - "github.com/pkg/errors" + "github.com/duke-git/lancet/v2/slice" "go.uber.org/multierr" + "github.com/apache/dubbo-admin/pkg/common/bizerror" "github.com/apache/dubbo-admin/pkg/config" "github.com/apache/dubbo-admin/pkg/config/console" "github.com/apache/dubbo-admin/pkg/config/diagnostics" "github.com/apache/dubbo-admin/pkg/config/discovery" "github.com/apache/dubbo-admin/pkg/config/engine" - "github.com/apache/dubbo-admin/pkg/config/mode" + "github.com/apache/dubbo-admin/pkg/config/log" + "github.com/apache/dubbo-admin/pkg/config/observability" "github.com/apache/dubbo-admin/pkg/config/store" ) type AdminConfig struct { config.BaseConfig - // Mode in which dubbo admin is running. Available values are: "test", "global", "zone" - Mode mode.Mode `json:"mode" envconfig:"DUBBO_MODE"` + // Log configuration + Log *log.Config `json:"log" yaml:"log"` // Diagnostics configuration - Diagnostics *diagnostics.Config `json:"diagnostics,omitempty"` + Diagnostics *diagnostics.Config `json:"diagnostics,omitempty" yaml:"diagnostics"` + // Observability configuration + Observability *observability.Config `json:"observability" yaml:"observability"` // Console configuration - Console *console.Config `json:"admin"` + Console *console.Config `json:"console" yaml:"console"` // Store configuration - Store *store.Config `json:"store"` + Store *store.Config `json:"store" yaml:"store"` // Discovery configuration - Discovery *discovery.Config `json:"discovery"` + Discovery []*discovery.Config `json:"discovery" yaml:"discovery"` // Engine configuration - Engine *engine.Config `json:"engine"` + Engine *engine.Config `json:"engine" yaml:"engine"` + // MCP configuration + MCP *MCPConfig `json:"mcp,omitempty" yaml:"mcp"` +} + +// MCPConfig MCP配置 +type MCPConfig struct { + // Enabled 是否启用MCP端点 + Enabled bool `json:"enabled" yaml:"enabled"` + // Path MCP端点路径，默认 /api/mcp + Path string `json:"path,omitempty" yaml:"path"` + // APIKey MCP API密钥，用于认证。如果为空则不需要认证 + APIKey string `json:"apiKey,omitempty" yaml:"apiKey"` } var _ = &AdminConfig{} -func (c *AdminConfig) Sanitize() { +var DefaultAdminConfig = func() AdminConfig { + return AdminConfig{ + Log: log.DefaultLogConfig(), + Store: store.DefaultStoreConfig(), + Engine: engine.DefaultResourceEngineConfig(), + Observability: observability.DefaultObservabilityConfig(), + Diagnostics: diagnostics.DefaultDiagnosticsConfig(), + Console: console.DefaultConsoleConfig(), + } +} + +func (c AdminConfig) Sanitize() { c.Engine.Sanitize() - c.Discovery.Sanitize() + for _, d := range c.Discovery { + d.Sanitize() + } c.Store.Sanitize() c.Console.Sanitize() + c.Observability.Sanitize() c.Diagnostics.Sanitize() + c.Log.Sanitize() } -func (c *AdminConfig) PostProcess() error { +func (c AdminConfig) PreProcess() error { + discoveryPreProcess := func() error { + for _, d := range c.Discovery { + if err := d.PreProcess(); err != nil { + return err + } + } + return nil + } + return multierr.Combine( + c.Engine.PreProcess(), + discoveryPreProcess(), + c.Store.PreProcess(), + c.Console.PreProcess(), + c.Observability.PreProcess(), + c.Diagnostics.PreProcess(), + c.Log.PreProcess(), + ) +} + +func (c AdminConfig) PostProcess() error { + discoveryPostProcess := func() error { + for _, d := range c.Discovery { + if err := d.PostProcess(); err != nil { + return err + } + } + return nil + } return multierr.Combine( c.Engine.PostProcess(), - c.Discovery.PostProcess(), + discoveryPostProcess(), c.Store.PostProcess(), c.Console.PostProcess(), + c.Observability.PostProcess(), c.Diagnostics.PostProcess(), + c.Log.PostProcess(), ) } -var DefaultAdminConfig = func() AdminConfig { - return AdminConfig{ - Mode: mode.Zone, - Store: store.DefaultStoreConfig(), - Engine: engine.DefaultResourceEngineConfig(), - Diagnostics: diagnostics.DefaultDiagnosticsConfig(), - Console: console.DefaultConsoleConfig(), - } -} - -func (c *AdminConfig) Validate() error { - if err := mode.ValidateMode(c.Mode); err != nil { - return errors.Wrap(err, "Mode Config validation failed") +func (c AdminConfig) Validate() error { + if c.Log == nil { + c.Log = log.DefaultLogConfig() + } else if err := c.Log.Validate(); err != nil { + return bizerror.Wrap(err, bizerror.ConfigError, "log config validation failed") Review Comment: `Validate()` 是值接收者，但这里会给 nil 字段赋默认值（例如 `c.Log = ...`）。值接收者会复制 config，导致默认值不会写回原始配置，后续可能仍为 nil 并引发 panic。建议把 `Validate/PreProcess/PostProcess/Sanitize` 改为指针接收者。 ########## pkg/mcp/tools/metrics.go: ########## @@ -0,0 +1,84 @@ +/* + * 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 ( + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/console/counter" + "github.com/apache/dubbo-admin/pkg/mcp/types" + "github.com/apache/dubbo-admin/pkg/mcp/registry" + meshresource "github.com/apache/dubbo-admin/pkg/core/resource/apis/mesh/v1alpha1" +) + +// MetricsRegistrar 集群工具注册器 +type MetricsRegistrar struct{} + +// RegisterTools 实现 ToolRegistrar 接口 +func (r *MetricsRegistrar) RegisterTools(reg *registry.Registry) { + reg.Register(types.ToolDef{ + Name: "get_cluster_info", + Description: "获取 Dubbo 集群基本信息，包括应用数、服务数、实例数等统计信息", + InputSchema: types.InputSchema{ + Type: "object", + Properties: map[string]types.PropertyDef{ + "mesh": { + Type: "string", + Description: "Mesh 名称，默认使用配置中的默认 mesh", + }, + }, + }, + Handler: GetClusterInfo, + }) +} + +// GetClusterInfo 获取集群基本信息 +func GetClusterInfo(ctx consolectx.Context, args map[string]any) (*types.ToolResult, error) { + mesh := GetMeshArg(ctx, args) + info := collectClusterInfo(ctx, mesh) + return JsonResult(info) +} + +// collectClusterInfo 收集集群信息 +func collectClusterInfo(ctx consolectx.Context, mesh string) map[string]any { + counterMgr := ctx.CounterManager() + if counterMgr == nil { + return map[string]any{ Review Comment: 这里调用了 `ctx.CounterManager()`，但 `pkg/console/context.Context` 接口并不包含该方法（当前接口只有 ResourceManager/ResourceStore/Config/AppContext）。这会导致编译失败。建议： - 要么扩展 console context 接口并在实现里提供 CounterManager； - 要么不要从 consolectx.Context 取 CounterManager，改为从 runtime 组件获取 counter manager（或通过依赖注入把 counter manager 传进来）。 ########## pkg/mcp/transport/http/handler.go: ########## @@ -0,0 +1,109 @@ +/* + * 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 http + +import ( + "encoding/json" + "io" + "net/http" + + "github.com/apache/dubbo-admin/pkg/mcp/core" +) + +// Handler HTTP请求处理器 +type Handler struct { + server *core.Server +} + +// NewHandler 创建HTTP处理器 +func NewHandler(server *core.Server) *Handler { + return &Handler{ + server: server, + } +} + +// ServeHTTP 实现http.Handler接口 +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // 设置CORS headers + h.setCORSHeaders(w) + + // 处理OPTIONS请求（CORS preflight） + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusOK) + return + } + + // 只接受POST请求 + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + // 读取请求体 + body, err := io.ReadAll(r.Body) + if err != nil { + h.sendError(w, nil, core.ErrCodeParseError, "Failed to read request body") + return + } + + // 解析JSON-RPC请求 + var req core.JSONRPCRequest + if err := json.Unmarshal(body, &req); err != nil { + h.sendError(w, nil, core.ErrCodeParseError, "Invalid JSON") + return + } + + // 处理请求并获取响应 + resp := h.server.HandleRequest(&req) + + // 发送响应 + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(resp) +} + +// setCORSHeaders 设置CORS headers +func (h *Handler) setCORSHeaders(w http.ResponseWriter) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "POST, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "Content-Type") Review Comment: CORS 只允许 `Content-Type`，但 MCP 在配置 APIKey 时需要 `Authorization: Bearer ...`。浏览器跨域请求会因预检不允许 Authorization header 而失败。建议把 `Authorization` 加入 `Access-Control-Allow-Headers`。 ########## pkg/mcp/transport/http/http_test.go: ########## @@ -0,0 +1,272 @@ +/* + * 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 http + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/apache/dubbo-admin/pkg/mcp/core" + "github.com/apache/dubbo-admin/pkg/mcp/registry" Review Comment: 这里导入了 `registry` 但文件中未使用；同时下面的并发测试使用了 `fmt.Sprintf`（但 `fmt` 未导入）。这会导致测试文件无法编译。建议移除未使用 import 并补齐 `fmt` import。 ########## pkg/console/component.go: ########## @@ -123,21 +159,101 @@ func (c *consoleWebServer) startHttpServer(errChan chan error) *http.Server { return server } +func (c *consoleWebServer) registerMCPEndpoints(coreRt runtime.Runtime, engine *gin.Engine) { + // 从runtime获取完整配置 + var cfg app.AdminConfig = coreRt.Config() + + // 检查MCP是否启用 + if cfg.MCP == nil || !cfg.MCP.Enabled { + return + } + + // 确定端点路径 + path := cfg.MCP.Path + if path == "" { + path = "/api/mcp" + } + + // 存储MCP路径和API Key供auth中间件使用 + c.mcpPath = path + c.mcpAPIKey = cfg.MCP.APIKey + + // 直接创建MCP服务器 + consoleCtx := consolectx.NewConsoleContext(coreRt) + server := mcpcore.NewServer("dubbo-admin", "1.0.0") + server.SetConsoleContext(consoleCtx) + + // 注册所有工具 + reg := server.GetRegistry() + reg.RegisterRegistrar(&mcp_tools.MetricsRegistrar{}) + reg.RegisterRegistrar(&mcp_tools.ResourceSearchRegistrar{}) + reg.RegisterRegistrar(&mcp_tools.ServiceRegistrar{}) + reg.RegisterRegistrar(&mcp_tools.DetailRegistrar{}) + reg.RegisterAll() + + // 创建HTTP处理器 + handler := mcphttp.NewHandler(server) + + // 注册路由 + engine.POST(path, func(ctx *gin.Context) { + handler.ServeHTTP(ctx.Writer, ctx.Request) + }) + + authStatus := "no-auth" + if c.mcpAPIKey != "" { + authStatus = "with-auth" + } + logger.Sugar().Infof("MCP endpoint registered at %s with %d tools (%s)", path, len(reg.List()), authStatus) +} + func (c *consoleWebServer) authMiddleware() gin.HandlerFunc { - return func(c *gin.Context) { + return func(ctx *gin.Context) { + requestPath := ctx.Request.URL.Path + // skip login api - requestPath := c.Request.URL.Path if strings.HasSuffix(requestPath, "/login") { - c.Next() + ctx.Next() + return + } + + // check MCP endpoint authentication + isMCPRequest := requestPath == "/api/mcp" || (c.mcpPath != "" && requestPath == c.mcpPath) + if isMCPRequest { + // 如果配置了 API Key，验证 Bearer Token + if c.mcpAPIKey != "" { + authHeader := ctx.GetHeader("Authorization") + if authHeader == "" { + ctx.JSON(http.StatusUnauthorized, gin.H{"error": "Missing Authorization header"}) + ctx.Abort() + return + } + // 检查 Bearer Token 格式 + if len(authHeader) < 7 || authHeader[:7] != "Bearer " { + ctx.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid Authorization header format. Use: Bearer <token>"}) + ctx.Abort() + return + } + token := authHeader[7:] + if token != c.mcpAPIKey { + ctx.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid API key"}) + ctx.Abort() Review Comment: API Key 校验使用了普通字符串比较 `token != c.mcpAPIKey`，在不受信任网络场景下可能泄露时序信息。建议用 `crypto/subtle.ConstantTimeCompare` 进行恒定时间比较。 ########## pkg/mcp/core/server.go: ########## @@ -0,0 +1,237 @@ +/* + * 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 core + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/mcp/registry" + "github.com/apache/dubbo-admin/pkg/mcp/types" + "github.com/gin-gonic/gin" +) + +// Server MCP 服务器 +type Server struct { + name string + version string + registry *registry.Registry + consoleContext consolectx.Context +} + +// NewServer 创建 MCP 服务器 +func NewServer(name, version string) *Server { + return &Server{ + name: name, + version: version, + registry: registry.NewRegistry(), + } +} + +// NewServerWithRegistry 使用指定 Registry 创建 MCP 服务器 +func NewServerWithRegistry(name, version string, reg *registry.Registry) *Server { + return &Server{ + name: name, + version: version, + registry: reg, + } +} + +// GetRegistry 获取工具注册表 +func (s *Server) GetRegistry() *registry.Registry { + return s.registry +} + +// SetConsoleContext 设置 console context +func (s *Server) SetConsoleContext(ctx consolectx.Context) { + s.consoleContext = ctx +} + +// GetConsoleContext 获取 console context +func (s *Server) GetConsoleContext() consolectx.Context { + return s.consoleContext +} + +// ==================== 请求处理 ==================== + +// HandleRequest 处理 JSON-RPC 请求（公开方法，供 transport 层使用） +func (s *Server) HandleRequest(req *JSONRPCRequest) *JSONRPCResponse { + return s.handleRequest(req) +} + +// HandleHTTP 处理 HTTP 请求 +func (s *Server) HandleHTTP(c *gin.Context) { + body, err := io.ReadAll(c.Request.Body) + if err != nil { + s.respondWithError(c, nil, ErrCodeParseError, "Parse error") + return + } + + var req JSONRPCRequest + if err := json.NewDecoder(bytes.NewReader(body)).Decode(&req); err != nil { + s.respondWithError(c, nil, ErrCodeParseError, "Parse error") + return + } + + c.JSON(http.StatusOK, s.handleRequest(&req)) +} + +// respondWithError 返回错误响应 +func (s *Server) respondWithError(c *gin.Context, id interface{}, code int, message string) { + c.JSON(http.StatusBadRequest, JSONRPCResponse{ + JSONRPC: JSONRPCVersion, Review Comment: `respondWithError` 接收了 `id` 参数但响应里没有设置 `ID` 字段，导致错误响应无法关联请求（并且参数当前未被使用）。建议在返回体中补上 `ID: id`（parse error 可保持 nil）。 ########## pkg/mcp/transport/http/http_test.go: ########## @@ -0,0 +1,272 @@ +/* + * 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 http + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/apache/dubbo-admin/pkg/mcp/core" + "github.com/apache/dubbo-admin/pkg/mcp/registry" +) + +// TestHTTPTransport 测试HTTP传输 +func TestHTTPTransport(t *testing.T) { + // 创建测试服务器 + server := core.NewServer("test-server", "1.0.0") + + // 注册测试工具 + reg := server.GetRegistry() + reg.Register(core.ToolDef{ + Name: "test_tool", + Description: "A test tool", + InputSchema: core.InputSchema{ + Type: "object", + Properties: map[string]core.PropertyDef{ + "message": { + Type: "string", + Description: "Test message", + }, + }, + }, + Handler: func(ctx interface{}, args map[string]any) (*core.ToolResult, error) { + msg, _ := args["message"].(string) + return core.NewTextResult("Echo: " + msg), nil + }, Review Comment: 该工具注册的 `Handler` 签名与 `core.ToolHandler` 不一致（应为 `func(consolectx.Context, map[string]any) ...`），并且 `core.NewTextResult` 需要 `isError` 参数。当前写法会导致编译失败。 ########## pkg/mcp/transport/http/http_test.go: ########## @@ -0,0 +1,272 @@ +/* + * 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 http + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/apache/dubbo-admin/pkg/mcp/core" + "github.com/apache/dubbo-admin/pkg/mcp/registry" +) + +// TestHTTPTransport 测试HTTP传输 +func TestHTTPTransport(t *testing.T) { + // 创建测试服务器 + server := core.NewServer("test-server", "1.0.0") + + // 注册测试工具 + reg := server.GetRegistry() + reg.Register(core.ToolDef{ + Name: "test_tool", + Description: "A test tool", + InputSchema: core.InputSchema{ + Type: "object", + Properties: map[string]core.PropertyDef{ + "message": { + Type: "string", + Description: "Test message", + }, + }, + }, + Handler: func(ctx interface{}, args map[string]any) (*core.ToolResult, error) { + msg, _ := args["message"].(string) + return core.NewTextResult("Echo: " + msg), nil + }, + }) + reg.RegisterAll() + + // 创建HTTP传输层 + transport := NewTransportWithConfig(server, &Config{ + Host: "127.0.0.1", + Port: 0, // 随机端口 + ReadTimeout: 5 * time.Second, + WriteTimeout: 5 * time.Second, + }) + + // 异步启动 + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + if err := transport.StartAsync(ctx); err != nil { + t.Fatalf("Failed to start transport: %v", err) + } + + // 等待服务器启动 + time.Sleep(100 * time.Millisecond) + + // 测试initialize请求 + t.Run("Initialize", func(t *testing.T) { + req := core.JSONRPCRequest{ + JSONRPC: core.JSONRPCVersion, + ID: "1", + Method: core.MethodInitialize, + } + + resp := makeRequest(t, transport, req) + if resp.Error != nil { + t.Fatalf("Initialize failed: %v", resp.Error) + } + + result, ok := resp.Result.(map[string]any) + if !ok { + t.Fatal("Invalid result type") + } + + serverInfo := result["serverInfo"].(map[string]any) + if serverInfo["name"] != "test-server" { + t.Errorf("Expected server name 'test-server', got '%v'", serverInfo["name"]) + } + }) + + // 测试tools/list请求 + t.Run("ToolsList", func(t *testing.T) { + req := core.JSONRPCRequest{ + JSONRPC: core.JSONRPCVersion, + ID: "2", + Method: core.MethodToolsList, + } + + resp := makeRequest(t, transport, req) + if resp.Error != nil { + t.Fatalf("Tools list failed: %v", resp.Error) + } + + result, ok := resp.Result.(map[string]any) + if !ok { + t.Fatal("Invalid result type") + } + + tools := result["tools"].([]any) + if len(tools) != 1 { + t.Errorf("Expected 1 tool, got %d", len(tools)) + } + }) + + // 测试tools/call请求 + t.Run("ToolsCall", func(t *testing.T) { + req := core.JSONRPCRequest{ + JSONRPC: core.JSONRPCVersion, + ID: "3", + Method: core.MethodToolsCall, + Params: map[string]any{ + "name": "test_tool", + "arguments": map[string]any{ + "message": "Hello, MCP!", + }, + }, + } + + resp := makeRequest(t, transport, req) + if resp.Error != nil { + t.Fatalf("Tools call failed: %v", resp.Error) + } + + result, ok := resp.Result.(map[string]any) + if !ok { + t.Fatal("Invalid result type") + } + + content := result["content"].([]any) + if len(content) == 0 { + t.Fatal("Empty content") + } + + firstContent := content[0].(map[string]any) + text := firstContent["text"].(string) + if text != "Echo: Hello, MCP!" { + t.Errorf("Expected 'Echo: Hello, MCP!', got '%s'", text) + } + }) + + // 关闭传输层 + if err := transport.Shutdown(); err != nil { + t.Fatalf("Failed to shutdown transport: %v", err) + } +} + +// makeRequest 发送请求到HTTP传输层 +func makeRequest(t *testing.T, transport *Transport, req core.JSONRPCRequest) core.JSONRPCResponse { + t.Helper() + + // 使用httptest直接测试handler + handler := transport.GetHandler() + body, _ := json.Marshal(req) + + httpReq := httptest.NewRequest("POST", "/mcp", bytes.NewReader(body)) + httpReq.Header.Set("Content-Type", "application/json") + + w := httptest.NewRecorder() + handler.ServeHTTP(w, httpReq) + + var resp core.JSONRPCResponse + if err := json.NewDecoder(w.Body).Decode(&resp); err != nil { + t.Fatalf("Failed to decode response: %v", err) + } + + return resp +} + +// TestHandlerCORS 测试CORS支持 +func TestHandlerCORS(t *testing.T) { + server := core.NewServer("test", "1.0.0") + handler := NewHandler(server) + + req := httptest.NewRequest("OPTIONS", "/mcp", nil) + req.Header.Set("Origin", "https://example.com";) + + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Errorf("Expected status 200, got %d", w.Code) + } + + corsHeader := w.Header().Get("Access-Control-Allow-Origin") + if corsHeader != "*" { + t.Errorf("Expected CORS origin '*', got '%s'", corsHeader) + } +} + +// TestConcurrentRequests 测试并发请求 +func TestConcurrentRequests(t *testing.T) { + server := core.NewServer("test", "1.0.0") + transport := NewTransportWithConfig(server, &Config{ + Host: "127.0.0.1", + Port: 0, + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + if err := transport.StartAsync(ctx); err != nil { + t.Fatalf("Failed to start transport: %v", err) + } + defer transport.Shutdown() + + time.Sleep(100 * time.Millisecond) + + // 并发发送10个请求 + const numRequests = 10 + errCh := make(chan error, numRequests) + + for i := 0; i < numRequests; i++ { + go func(id int) { + req := core.JSONRPCRequest{ + JSONRPC: core.JSONRPCVersion, + ID: fmt.Sprintf("req-%d", id), + Method: core.MethodInitialize, + } + _ = makeRequest(t, transport, req) + errCh <- nil + }(i) + } + + // 等待所有请求完成 + for i := 0; i < numRequests; i++ { + if err := <-errCh; err != nil { + t.Errorf("Request failed: %v", err) + } + } +} + +// TestSSETransport 测试SSE传输 +func TestSSETransport(t *testing.T) { + server := core.NewServer("test", "1.0.0") + sseTransport := NewSSETransport(server) + + // 测试SSE连接 + t.Run("SSEConnection", func(t *testing.T) { + req := httptest.NewRequest("GET", "/sse", nil) + w := httptest.NewRecorder() + + // SSE需要flusher支持，httptest.NewRecorder实现了Flusher接口 + sseTransport.HandleSSE(w, req) + Review Comment: `HandleSSE` 当前实现会阻塞等待断开连接，因此在单测中直接调用会永久挂起（此处没有任何 cancel/断开条件）。建议把 `HandleSSE` 放到 goroutine 并通过 `req.Context()` 取消，或重构为可测试的非阻塞逻辑。 ########## pkg/mcp/transport/http/http.go: ########## @@ -0,0 +1,168 @@ +/* + * 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 http + +import ( + "context" + "fmt" + "net/http" + "sync" + "time" + + "github.com/apache/dubbo-admin/pkg/mcp/core" +) + +// Transport HTTP传输层 +type Transport struct { + server *core.Server + httpServer *http.Server + handler *Handler + mu sync.RWMutex + started bool +} + +// Config HTTP传输配置 +type Config struct { + Host string + Port int + ReadTimeout time.Duration + WriteTimeout time.Duration + ShutdownTimeout time.Duration +} + +// DefaultConfig 返回默认配置 +func DefaultConfig() *Config { + return &Config{ + Host: "0.0.0.0", + Port: 8080, + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + ShutdownTimeout: 10 * time.Second, + } +} + +// NewTransport 创建HTTP传输层 +func NewTransport(server *core.Server) *Transport { + return NewTransportWithConfig(server, DefaultConfig()) +} + +// NewTransportWithConfig 使用指定配置创建HTTP传输层 +func NewTransportWithConfig(server *core.Server, cfg *Config) *Transport { + handler := NewHandler(server) + + addr := fmt.Sprintf("%s:%d", cfg.Host, cfg.Port) + httpServer := &http.Server{ + Addr: addr, + Handler: handler, + ReadTimeout: cfg.ReadTimeout, + WriteTimeout: cfg.WriteTimeout, + } + + return &Transport{ + server: server, + httpServer: httpServer, + handler: handler, + } +} + +// Start 启动HTTP服务器（阻塞运行） +func (t *Transport) Start(ctx context.Context) error { + t.mu.Lock() + if t.started { + t.mu.Unlock() + return fmt.Errorf("transport already started") + } + t.started = true + t.mu.Unlock() + + // 启动服务器 + errCh := make(chan error, 1) + go func() { + if err := t.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + errCh <- err + } + }() + + // 等待上下文取消或错误 + select { + case <-ctx.Done(): + return t.Shutdown() + case err := <-errCh: + return err + } +} + +// StartAsync 异步启动HTTP服务器 +func (t *Transport) StartAsync(ctx context.Context) error { + t.mu.Lock() + if t.started { + t.mu.Unlock() + return fmt.Errorf("transport already started") + } + t.started = true + t.mu.Unlock() + + go func() { + if err := t.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + // 记录错误但不退出 + fmt.Printf("HTTP server error: %v\n", err) + } Review Comment: `StartAsync` 在 goroutine 中用 `fmt.Printf` 输出错误，会绕过项目 logger 且不便于统一日志收集。建议改用项目 logger 或将错误回传给调用方。 ########## pkg/mcp/component.go: ########## @@ -0,0 +1,150 @@ +/* + * 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 mcp + +import ( + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/config/app" + "github.com/apache/dubbo-admin/pkg/core/runtime" + "github.com/apache/dubbo-admin/pkg/mcp/core" + "github.com/apache/dubbo-admin/pkg/mcp/tools" + "github.com/apache/dubbo-admin/pkg/mcp/transport/http" +) + +const ( + // ComponentType MCP组件类型 + ComponentType = runtime.ComponentType("mcp") +) + +// Component MCP组件，集成到admin服务中 +type Component struct { + server *core.Server + consoleCtx consolectx.Context + cfg *app.MCPConfig +} + +func init() { + runtime.RegisterComponent(&Component{}) +} + +// Type 返回组件类型 +func (c *Component) Type() runtime.ComponentType { + return ComponentType +} + +// Order 返回组件启动顺序 +func (c *Component) Order() int { + return 999 // 在Console之后启动 +} + +// RequiredDependencies 返回依赖的组件 +func (c *Component) RequiredDependencies() []runtime.ComponentType { + return []runtime.ComponentType{ + runtime.Console, // 依赖Console（需要runtime） + } +} + +// Init 初始化MCP组件 +func (c *Component) Init(ctx runtime.BuilderContext) error { + cfg := ctx.Config() + + // 从admin配置中获取MCP配置 + if cfg.MCP == nil { + // MCP未配置，使用默认配置（禁用） + c.cfg = &app.MCPConfig{ + Enabled: false, + Path: "/api/mcp", + } + return nil + } + + c.cfg = cfg.MCP + + // 如果未启用，直接返回 + if !c.cfg.Enabled { + return nil + } + + return nil +} + +// Start 启动MCP组件 +func (c *Component) Start(coreRt runtime.Runtime, stop <-chan struct{}) error { + // 如果未启用，直接返回 + if c.cfg == nil || !c.cfg.Enabled { + <-stop + return nil + } + + // 获取console context（由Console组件创建） + c.consoleCtx = consolectx.NewConsoleContext(coreRt) + + // 创建MCP服务器 + c.server = core.NewServer("dubbo-admin", "1.0.0") + c.server.SetConsoleContext(c.consoleCtx) + + // 注册所有工具 + reg := c.server.GetRegistry() + reg.RegisterRegistrar(&tools.MetricsRegistrar{}) + reg.RegisterRegistrar(&tools.ResourceSearchRegistrar{}) + reg.RegisterRegistrar(&tools.ServiceRegistrar{}) + reg.RegisterRegistrar(&tools.DetailRegistrar{}) + reg.RegisterAll() + + // 获取console的HTTP引擎并注册MCP路由 + consoleComp, err := coreRt.GetComponent(runtime.Console) + if err != nil { + // Console未启动，无法注册MCP路由 + <-stop + return nil + } + + // 通过runtime获取gin.Engine + // 这里需要一种方式来访问console组件的gin.Engine + // 暂时使用全局注册方式 + RegisterMCPRoutes(c.server, c.cfg.Path) + + _ = consoleComp // 暂时忽略，实际使用时需要访问console的gin.Engine + + // 等待停止信号 + <-stop + + return nil +} + +// RegisterMCPRoutes 注册MCP路由到gin引擎 +// 这个函数应该在console路由初始化后调用 +// 注意：由于这个函数使用gin.Default()，它创建了一个新的引擎实例 +// 在实际使用中，应该由console组件调用并注册到它自己的引擎上 +func RegisterMCPRoutes(server *core.Server, path string) { + if path == "" { + path = "/api/mcp" + } + + handler := http.NewHandler(server) + + // 注册MCP端点（不需要认证） + // 注意：这只是示例代码，实际注册应该在console组件中完成 + _ = handler // 避免未使用警告 Review Comment: `RegisterMCPRoutes` 当前只是构造了 handler 然后直接丢弃（`_ = handler`），不会把路由注册到任何 gin.Engine 上，因此该函数实际是 no-op。建议删除该占位实现或改为接收并注册到真实的 gin.Engine。 ########## pkg/mcp/component.go: ########## @@ -0,0 +1,150 @@ +/* + * 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 mcp + +import ( + consolectx "github.com/apache/dubbo-admin/pkg/console/context" + "github.com/apache/dubbo-admin/pkg/config/app" + "github.com/apache/dubbo-admin/pkg/core/runtime" + "github.com/apache/dubbo-admin/pkg/mcp/core" + "github.com/apache/dubbo-admin/pkg/mcp/tools" + "github.com/apache/dubbo-admin/pkg/mcp/transport/http" +) + +const ( + // ComponentType MCP组件类型 + ComponentType = runtime.ComponentType("mcp") +) + +// Component MCP组件，集成到admin服务中 +type Component struct { + server *core.Server + consoleCtx consolectx.Context + cfg *app.MCPConfig +} + +func init() { + runtime.RegisterComponent(&Component{}) +} + +// Type 返回组件类型 +func (c *Component) Type() runtime.ComponentType { + return ComponentType +} + +// Order 返回组件启动顺序 +func (c *Component) Order() int { + return 999 // 在Console之后启动 +} + +// RequiredDependencies 返回依赖的组件 +func (c *Component) RequiredDependencies() []runtime.ComponentType { + return []runtime.ComponentType{ + runtime.Console, // 依赖Console（需要runtime） + } +} + +// Init 初始化MCP组件 +func (c *Component) Init(ctx runtime.BuilderContext) error { + cfg := ctx.Config() + + // 从admin配置中获取MCP配置 + if cfg.MCP == nil { + // MCP未配置，使用默认配置（禁用） + c.cfg = &app.MCPConfig{ + Enabled: false, + Path: "/api/mcp", + } + return nil + } + + c.cfg = cfg.MCP + + // 如果未启用，直接返回 + if !c.cfg.Enabled { + return nil + } + + return nil +} + +// Start 启动MCP组件 +func (c *Component) Start(coreRt runtime.Runtime, stop <-chan struct{}) error { + // 如果未启用，直接返回 + if c.cfg == nil || !c.cfg.Enabled { + <-stop + return nil + } + + // 获取console context（由Console组件创建） + c.consoleCtx = consolectx.NewConsoleContext(coreRt) + + // 创建MCP服务器 + c.server = core.NewServer("dubbo-admin", "1.0.0") + c.server.SetConsoleContext(c.consoleCtx) + + // 注册所有工具 + reg := c.server.GetRegistry() + reg.RegisterRegistrar(&tools.MetricsRegistrar{}) + reg.RegisterRegistrar(&tools.ResourceSearchRegistrar{}) + reg.RegisterRegistrar(&tools.ServiceRegistrar{}) + reg.RegisterRegistrar(&tools.DetailRegistrar{}) + reg.RegisterAll() + + // 获取console的HTTP引擎并注册MCP路由 + consoleComp, err := coreRt.GetComponent(runtime.Console) + if err != nil { + // Console未启动，无法注册MCP路由 + <-stop + return nil + } + + // 通过runtime获取gin.Engine + // 这里需要一种方式来访问console组件的gin.Engine + // 暂时使用全局注册方式 + RegisterMCPRoutes(c.server, c.cfg.Path) + Review Comment: 这里调用 `RegisterMCPRoutes(...)` 但没有把路由注册到 console 的 gin.Engine 上（`consoleComp` 也被忽略）。同时 console 组件里已经实现了 MCP 路由注册逻辑，容易造成两套集成方式并存且其中一套无效。建议统一集成入口，避免重复/混淆。 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
