Similarityoung commented on code in PR #953:
URL: https://github.com/apache/dubbo-go-pixiu/pull/953#discussion_r3487659072
##########
pkg/adapter/mcpserver/registrycenter.go:
##########
@@ -212,6 +184,53 @@ func (a *Adapter) Apply() error {
return nil
}
+func (a *Adapter) applyServerConfigEvent(reconciler *endpointReconciler, sink
mcpserver.ServerPublicationSink, registryName, serverId string, cfg
*model.McpServerConfig) {
+ if serverId == "" {
+ serverId = "default"
+ }
+
+ // Apply catalog and endpoints through one desired-state publication
path. If
+ // no runtime target is bound, skip the entire update so authorization
catalog
+ // and cluster endpoints cannot diverge.
+ if sink == nil {
+ logger.Infof("[dubbo-go-pixiu] mcp adapter update received from
server %s without a bound runtime publication sink", serverId)
+ return
+ }
+ runtimeID := sink.RuntimeID()
+ if runtimeID == "" {
+ logger.Errorf("[dubbo-go-pixiu] mcp adapter update received
from server %s but publication sink has no runtime id", serverId)
+ return
+ }
+ if err := reconciler.ValidateServerConfig(runtimeID, registryName,
serverId, cfg); err != nil {
+ logger.Errorf("[dubbo-go-pixiu] mcp adapter validate endpoints
for server %s error: %v", serverId, err)
+ return
+ }
+ if err := sink.ApplyMcpServerConfigByServer(serverId, cfg); err != nil {
Review Comment:
这里已经支持把 `cfg == nil` 传给 dynamic consumer 做删除，但 Nacos controller 在 server
从列表消失时只 `CancelListenToServer`，没有发布 `onChange(serverId, nil)` tombstone。因此真实
server 下线路径仍不会清理 catalog 和 endpoints。建议 controller 在移除 watched server 时显式发送
tombstone，并补 server 发布 tools 后从 registry 消失时 catalog/endpoints 都被清理的测试。
##########
pkg/filter/mcp/mcpserver/runtime.go:
##########
@@ -0,0 +1,148 @@
+/*
+ * 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 mcpserver
+
+import (
+ "errors"
+ "fmt"
+ "sync"
+ "sync/atomic"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/model"
+)
+
+var (
+ runtimeIndexMu sync.RWMutex
+ runtimeIndex = map[string]*RuntimeState{}
+ runtimeSeq uint64
+ resetTestGlobals func()
+)
+
+var ErrDynamicConsumerUnavailable = errors.New("mcp dynamic consumer
unavailable")
+var ErrDynamicConsumerAmbiguous = errors.New("multiple mcp dynamic consumers
registered")
+
+// ServerPublicationSink is the dynamic registry publication target owned by
one
+// MCP runtime. Registry adapters bind to this sink once instead of
discovering a
+// runtime on every event.
+type ServerPublicationSink interface {
+ RuntimeID() string
+ ApplyMcpServerConfigByServer(serverId string, cfg
*model.McpServerConfig) error
+}
+
+func registerRuntime(runtime *RuntimeState) string {
+ if runtime == nil {
+ return ""
+ }
+ id := fmt.Sprintf("mcp-runtime-%d", atomic.AddUint64(&runtimeSeq, 1))
+ runtimeIndexMu.Lock()
+ runtimeIndex[id] = runtime
+ runtimeIndexMu.Unlock()
+ if runtime.dynamic != nil {
+ runtime.dynamic.setRuntimeID(id)
+ }
+ return id
+}
+
+func unregisterRuntime(id string) {
+ if id == "" {
+ return
+ }
+ runtimeIndexMu.Lock()
+ runtime := runtimeIndex[id]
+ delete(runtimeIndex, id)
+ runtimeIndexMu.Unlock()
+ if runtime != nil && runtime.dynamic != nil {
+ runtime.dynamic.setRuntimeID("")
+ }
+}
+
+// Stop releases all runtime-owned state. It does not affect other MCP filters.
+func (r *RuntimeState) Stop() {
+ if r == nil {
+ return
+ }
+ unregisterRuntime(r.id)
+ if r.sessionManager != nil {
+ r.sessionManager.Stop()
+ }
+ if r.plans != nil {
+ r.plans.Stop()
+ }
+}
+
+// DynamicConsumerForSingleRuntime returns the only registered dynamic
consumer.
+// Registry-center callbacks are process-global today, so multiple MCP filters
+// must be treated as ambiguous instead of guessing which runtime should
update.
+func DynamicConsumerForSingleRuntime() (*DynamicConsumer, error) {
+ runtimeIndexMu.RLock()
+ defer runtimeIndexMu.RUnlock()
+ switch len(runtimeIndex) {
+ case 0:
+ return nil, ErrDynamicConsumerUnavailable
+ case 1:
+ for _, runtime := range runtimeIndex {
+ if runtime.dynamic == nil {
+ return nil, ErrDynamicConsumerUnavailable
+ }
+ return runtime.dynamic, nil
+ }
+ default:
+ return nil, ErrDynamicConsumerAmbiguous
Review Comment:
多 MCP runtime 时这里直接返回 ambiguous，adapter 又没有配置维度来绑定目标 runtime；只要进程里有多个 MCP
server filter，动态 registry publication 就整体不可用。建议增加显式绑定规则（例如
adapter/filter/runtime id 或 endpoint 维度），或者在配置层限制并启动失败，避免运行时静默跳过动态更新。
##########
pkg/filter/mcp/mcpserver/dynamic.go:
##########
@@ -70,93 +87,139 @@ func NewDynamicConsumer(reg *ToolRegistry, sm
*transport.SessionManager, sseHand
}
}
-// ApplyMcpServerConfigByServer applies configuration by server ID
+func (d *DynamicConsumer) RuntimeID() string {
+ d.mu.RLock()
+ defer d.mu.RUnlock()
+ return d.runtimeID
+}
+
+func (d *DynamicConsumer) setRuntimeID(id string) {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ d.runtimeID = id
+}
+
+func (d *DynamicConsumer) SetGovernance(selector router.ToolSelector, plans
*router.SessionPlanStore) {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ d.selector = selector
+ d.plans = plans
+ d.governance = selector != nil && plans != nil
+}
+
+// ApplyMcpServerConfigByServer applies a dynamic tool catalog update by server
+// ID. Router configuration is intentionally not applied on this path; a
dynamic
+// payload that contains a router section is rejected before mutating tools so
+// the runtime cannot enter a tools=new/router=old partial state.
func (d *DynamicConsumer) ApplyMcpServerConfigByServer(serverId string, cfg
*model.McpServerConfig) error {
if cfg == nil {
return d.removeServerConfig(serverId)
}
+ if err := d.validateDynamicConfig(cfg); err != nil {
+ return err
+ }
- d.mu.Lock()
- defer d.mu.Unlock()
+ result, err := d.applyServerTools(serverId, cfg.Tools)
+ if err != nil {
+ return err
+ }
+ if d.logSkippedServerConfigApply(serverId, result) {
+ return nil
+ }
- // 1. Calculate new configuration fingerprint
- fingerprint := d.calculateFingerprint(cfg.Tools)
+ logger.Infof("[dubbo-go-pixiu] mcp server %s config applied: %d tools,
total servers: %d, merged tools: %d",
+ serverId, len(cfg.Tools), result.ServerCount,
result.MergedCount)
- // 2. Check if the server's configuration really needs to be updated
- if existingConfig, exists := d.serverConfigs[serverId]; exists {
- if existingConfig.Fingerprint == fingerprint {
- logger.Debugf("[dubbo-go-pixiu] mcp server %s config
unchanged (fp=%s), skipped", serverId, fingerprint)
- return nil
- }
+ d.notifyToolsListChanged()
+
+ return nil
+}
+
+func (d *DynamicConsumer) validateDynamicConfig(cfg *model.McpServerConfig)
error {
+ if cfg.Router != nil {
+ return errDynamicRouterUpdateUnsupported
}
+ if !d.governanceEnabled() {
+ return nil
+ }
+ if err := router.ValidateTools(cfg.Tools); err != nil {
+ return fmt.Errorf("invalid mcp tool router metadata: %w", err)
+ }
+ return nil
+}
- // 3. Debounce check (based on this server's configuration change time)
+func (d *DynamicConsumer) applyServerTools(serverId string, tools
[]model.ToolConfig) (serverConfigApplyResult, error) {
+ result := serverConfigApplyResult{Fingerprint:
d.calculateFingerprint(tools)}
now := time.Now()
- if existingConfig, exists := d.serverConfigs[serverId]; exists {
- // Skip only if this server is within debounce time
- if !existingConfig.LastApplied.IsZero() &&
now.Sub(existingConfig.LastApplied) < d.debounceTime {
- logger.Debugf("[dubbo-go-pixiu] mcp server %s debounce
active (elapsed=%v), skipped", serverId, now.Sub(existingConfig.LastApplied))
- return nil
- }
+
+ d.mu.Lock()
+ defer d.mu.Unlock()
+
+ existingConfig := d.serverConfigs[serverId]
+ if existingConfig != nil && existingConfig.Fingerprint ==
result.Fingerprint {
+ result.SkippedReason = serverConfigApplySkippedUnchanged
+ return result, nil
+ }
+ if elapsed, ok := d.serverConfigDebounceElapsed(existingConfig, now);
ok {
Review Comment:
这个 debounce 会跳过同一 server 在 500ms 内的第二次更新，即使 fingerprint
已经不同。`ApplyMcpServerConfigByServer` 对 caller 仍返回 nil，adapter 后续会继续按新 cfg
reconcile endpoints，导致 catalog 还是旧工具、cluster endpoints 已经是新配置。建议只跳过 fingerprint
相同的重复事件；不同 fingerprint 要么立即提交，要么 debounce 到最新 desired state 后统一提交，不能丢更新。
##########
pkg/adapter/mcpserver/registrycenter.go:
##########
@@ -157,47 +158,18 @@ func (a *Adapter) Apply() error {
continue
}
+ reconciler := a.endpoints
+ if reconciler == nil {
+ reconciler =
newEndpointReconciler(clusterManagerEndpointSink{})
+ a.endpoints = reconciler
+ }
+ registryName := k
+ sink, err := a.bindPublicationSink()
+ if err != nil {
+ logger.Infof("[dubbo-go-pixiu] mcp adapter registry %s
has no bound runtime publication sink: %v", registryName, err)
+ }
onChange := func(serverId string, cfg *model.McpServerConfig) {
- if cfg == nil {
- return
- }
-
- if serverId == "" {
- serverId = "default"
- }
-
- // 1) apply tools dynamically to registry for filter
usage
- if dc := mcpserver.GetOrInitDynamicConsumer(); dc !=
nil {
- if err :=
dc.ApplyMcpServerConfigByServer(serverId, cfg); err != nil {
- logger.Errorf("[dubbo-go-pixiu] mcp
adapter apply server %s config error: %v", serverId, err)
- }
- } else {
- logger.Infof("[dubbo-go-pixiu] mcp adapter
update received from server %s: tools=%d", serverId, len(cfg.Tools))
- }
- // 2) register endpoint for each tool using BackendURL
(host:port) into cluster named by tool.Name
- for _, tool := range cfg.Tools {
- if tool.BackendURL == "" {
- continue
- }
- result, err :=
util.ParseHostPortFromURL(tool.BackendURL)
- if err != nil {
- logger.Errorf("[dubbo-go-pixiu] mcp
adapter failed to parse BackendURL '%s' for tool '%s': %v",
- tool.BackendURL, tool.Name, err)
- continue
- }
- if result.UsedFallback {
- logger.Warnf("[dubbo-go-pixiu] mcp
adapter using fallback for tool '%s' with BackendURL '%s': %s",
- tool.Name, tool.BackendURL,
result.FallbackInfo)
- }
- endpointID := result.Host + ":" +
strconv.Itoa(result.Port)
-
server.GetClusterManager().SetEndpoint(tool.Cluster, &model.Endpoint{
- ID: endpointID,
- Address: model.SocketAddress{
- Address: result.Host,
- Port: result.Port,
- },
- })
- }
+ a.applyServerConfigEvent(reconciler, sink,
registryName, serverId, cfg)
Review Comment:
这里的 `sink` 是在 `Adapter.Apply()` 时绑定并被 `onChange` 闭包捕获的，但实际启动顺序里 adapter
manager 早于 listener/filter manager 初始化；MCP runtime 要到 HTTP connection manager
加载 MCP filter 后才注册。这样常规启动路径很容易先捕获到 `nil`，之后所有 registry event 都会被
`applyServerConfigEvent(..., sink=nil, ...)` 跳过。建议在每次 event 里懒绑定 publication
sink，或者调整启动依赖并加一个 adapter 先 Apply、filter 后 Apply、随后 event 仍可发布
catalog/endpoints 的集成测试。
--
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]
