fix: refactored mcp server auto discovery logic and fix some issue (#2382)

Co-authored-by: johnlanni <zty98751@alibaba-inc.com>

registry/nacos/v2/watcher.go

@@ -16,11 +16,14 @@ package v2
 
 import (
 	"errors"
+	"fmt"
+	"net"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/golang/protobuf/ptypes/wrappers"
 	"github.com/nacos-group/nacos-sdk-go/v2/clients"
 	"github.com/nacos-group/nacos-sdk-go/v2/clients/naming_client"
 	"github.com/nacos-group/nacos-sdk-go/v2/common/constant"
@@ -32,9 +35,11 @@ import (
 
 	apiv1 "github.com/alibaba/higress/api/networking/v1"
 	"github.com/alibaba/higress/pkg/common"
+	"github.com/alibaba/higress/registry"
 	provider "github.com/alibaba/higress/registry"
 	"github.com/alibaba/higress/registry/memory"
 	"github.com/alibaba/higress/registry/nacos/address"
+	"github.com/alibaba/higress/registry/nacos/mcpserver"
 )
 
 const (
@@ -68,6 +73,9 @@ type watcher struct {
 	updateCacheWhenEmpty bool
 	nacosClientConfig    *constant.ClientConfig
 	authOption           provider.AuthOption
+	namespace            string
+	clusterId            string
+	mcpWatcher           provider.Watcher
 }
 
 type WatcherOption func(w *watcher)
@@ -88,6 +96,45 @@ func NewWatcher(cache memory.Cache, opts ...WatcherOption) (provider.Watcher, er
 		opt(w)
 	}
 
+	if w.EnableMCPServer != nil && w.EnableMCPServer.GetValue() {
+		if w.Type != string(registry.Nacos3) {
+			log.Errorf("can not create mcpWatcher for nacos 2.x type, required nacos 3.x")
+		} else {
+			mcpWatcher, err := mcpserver.NewWatcher(
+				cache,
+				mcpserver.WithType(w.Type),
+				mcpserver.WithName(w.Name),
+				mcpserver.WithNacosAddressServer(w.NacosAddressServer),
+				mcpserver.WithDomain(w.Domain),
+				mcpserver.WithPort(w.Port),
+				mcpserver.WithNacosNamespaceId(w.NacosNamespaceId),
+				mcpserver.WithNacosNamespace(w.NacosNamespace),
+				mcpserver.WithNacosGroups(w.NacosGroups),
+				mcpserver.WithNacosAccessKey(w.NacosAccessKey),
+				mcpserver.WithNacosSecretKey(w.NacosSecretKey),
+				mcpserver.WithNacosRefreshInterval(w.NacosRefreshInterval),
+				mcpserver.WithMcpExportDomains(w.McpServerExportDomains),
+				mcpserver.WithMcpBaseUrl(w.McpServerBaseUrl),
+				mcpserver.WithEnableMcpServer(w.EnableMCPServer),
+				mcpserver.WithClusterId(w.clusterId),
+				mcpserver.WithNamespace(w.namespace),
+				mcpserver.WithAuthOption(w.authOption),
+			)
+			if err != nil {
+				return nil, fmt.Errorf("can not create mcp server watcher, err:%v", err)
+			}
+			var once sync.Once
+			mcpWatcher.ReadyHandler(func(ready bool) {
+				once.Do(func() {
+					if ready {
+						log.Infof("Registry mcp Watcher is ready, type:%s, name:%s", w.Type, w.Name)
+					}
+				})
+			})
+			w.mcpWatcher = mcpWatcher
+		}
+	}
+
 	if w.NacosNamespace == "" {
 		w.NacosNamespace = w.NacosNamespaceId
 	}
@@ -233,15 +280,51 @@ func WithAuthOption(authOption provider.AuthOption) WatcherOption {
 	}
 }
 
+func WithMcpExportDomains(exportDomains []string) WatcherOption {
+	return func(w *watcher) {
+		w.McpServerExportDomains = exportDomains
+	}
+}
+
+func WithMcpBaseUrl(url string) WatcherOption {
+	return func(w *watcher) {
+		w.McpServerBaseUrl = url
+	}
+}
+
+func WithEnableMcpServer(enable *wrappers.BoolValue) WatcherOption {
+	return func(w *watcher) {
+		w.EnableMCPServer = enable
+	}
+}
+
+func WithNamespace(ns string) WatcherOption {
+	return func(w *watcher) {
+		w.namespace = ns
+	}
+}
+
+func WithClusterId(id string) WatcherOption {
+	return func(w *watcher) {
+		w.clusterId = id
+	}
+}
+
 func (w *watcher) Run() {
 	ticker := time.NewTicker(time.Duration(w.NacosRefreshInterval))
 	defer ticker.Stop()
 	w.Status = provider.ProbeWatcherStatus(w.Domain, strconv.FormatUint(uint64(w.Port), 10))
+	if w.mcpWatcher != nil {
+		w.mcpWatcher.AppendServiceUpdateHandler(w.UpdateService)
+		go w.mcpWatcher.Run()
+	}
 	err := w.fetchAllServices()
 	if err != nil {
 		log.Errorf("first fetch services failed, err:%v", err)
 	} else {
-		w.Ready(true)
+		if w.mcpWatcherReady() {
+			w.Ready(true)
+		}
 	}
 	for {
 		select {
@@ -250,7 +333,9 @@ func (w *watcher) Run() {
 			if err != nil {
 				log.Errorf("fetch services failed, err:%v", err)
 			} else {
-				w.Ready(true)
+				if w.mcpWatcherReady() {
+					w.Ready(true)
+				}
 			}
 		case <-w.stop:
 			return
@@ -258,6 +343,10 @@ func (w *watcher) Run() {
 	}
 }
 
+func (w *watcher) mcpWatcherReady() bool {
+	return w.mcpWatcher == nil || w.mcpWatcher.IsReady()
+}
+
 func (w *watcher) updateNacosClient() {
 	for {
 		select {
@@ -438,6 +527,7 @@ func (w *watcher) getSubscribeCallback(groupName string, serviceName string) fun
 func (w *watcher) generateServiceEntry(host string, services []model.Instance) *v1alpha3.ServiceEntry {
 	portList := make([]*v1alpha3.ServicePort, 0)
 	endpoints := make([]*v1alpha3.WorkloadEntry, 0)
+	isDnsService := false
 
 	for _, service := range services {
 		protocol := common.HTTP
@@ -452,6 +542,9 @@ func (w *watcher) generateServiceEntry(host string, services []model.Instance) *
 		if len(portList) == 0 {
 			portList = append(portList, port)
 		}
+		if !isValidIP(service.Ip) {
+			isDnsService = true
+		}
 		endpoint := &v1alpha3.WorkloadEntry{
 			Address: service.Ip,
 			Ports:   map[string]uint32{port.Protocol: port.Number},
@@ -460,11 +553,15 @@ func (w *watcher) generateServiceEntry(host string, services []model.Instance) *
 		endpoints = append(endpoints, endpoint)
 	}
 
+	resolution := v1alpha3.ServiceEntry_STATIC
+	if isDnsService {
+		resolution = v1alpha3.ServiceEntry_DNS
+	}
 	se := &v1alpha3.ServiceEntry{
 		Hosts:      []string{host},
 		Ports:      portList,
 		Location:   v1alpha3.ServiceEntry_MESH_INTERNAL,
-		Resolution: v1alpha3.ServiceEntry_STATIC,
+		Resolution: resolution,
 		Endpoints:  endpoints,
 	}
 
@@ -477,6 +574,9 @@ func (w *watcher) Stop() {
 	if w.addrProvider != nil {
 		w.addrProvider.Stop()
 	}
+	if w.mcpWatcher != nil {
+		w.mcpWatcher.Stop()
+	}
 	for key := range w.WatchingServices {
 		s := strings.Split(key, DefaultJoiner)
 		err := w.unsubscribe(s[0], s[1])
@@ -523,3 +623,8 @@ func shouldSubscribe(serviceName string) bool {
 
 	return true
 }
+
+func isValidIP(ipStr string) bool {
+	ip := net.ParseIP(ipStr)
+	return ip != nil
+}