feat(metadata): add concurrency safety for application-level metadata state (#3353)

metadata/info/metadata_info.go

@@ -21,6 +21,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 )
 
 import (
@@ -63,6 +64,7 @@ type MetadataInfo struct {
 	Services              map[string]*ServiceInfo  `json:"services,omitempty" hessian:"services"`
 	exportedServiceURLs   map[string][]*common.URL `hessian:"-"` // server exported service urls
 	subscribedServiceURLs map[string][]*common.URL `hessian:"-"` // client subscribed service urls
+	mu                    sync.RWMutex             `json:"-" hessian:"-"`
 }
 
 func NewAppMetadataInfo(app string) *MetadataInfo {
@@ -95,6 +97,9 @@ func (info *MetadataInfo) JavaClassName() string {
 
 // AddService add provider service info to MetadataInfo
 func (info *MetadataInfo) AddService(url *common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
 	service := NewServiceInfoWithURL(url)
 	info.Services[service.GetMatchKey()] = service
 	addUrl(info.exportedServiceURLs, url)
@@ -131,22 +136,34 @@ func deleteItem(slice []*common.URL, index int) []*common.URL {
 }
 
 func (info *MetadataInfo) RemoveService(url *common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
 	service := NewServiceInfoWithURL(url)
 	delete(info.Services, service.GetMatchKey())
 	removeUrl(info.exportedServiceURLs, url)
 }
 
 // AddSubscribeURL client subscribe a service url
 func (info *MetadataInfo) AddSubscribeURL(url *common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
 	addUrl(info.subscribedServiceURLs, url)
 }
 
 // RemoveSubscribeURL client unsubscribe a service url
 func (info *MetadataInfo) RemoveSubscribeURL(url *common.URL) {
+	info.mu.Lock()
+	defer info.mu.Unlock()
+
 	removeUrl(info.subscribedServiceURLs, url)
 }
 
 func (info *MetadataInfo) GetExportedServiceURLs() []*common.URL {
+	info.mu.RLock()
+	defer info.mu.RUnlock()
+
 	res := make([]*common.URL, 0)
 	for _, urls := range info.exportedServiceURLs {
 		res = append(res, urls...)
@@ -155,13 +172,28 @@ func (info *MetadataInfo) GetExportedServiceURLs() []*common.URL {
 }
 
 func (info *MetadataInfo) GetSubscribedURLs() []*common.URL {
+	info.mu.RLock()
+	defer info.mu.RUnlock()
+
 	res := make([]*common.URL, 0)
 	for _, urls := range info.subscribedServiceURLs {
 		res = append(res, urls...)
 	}
 	return res
 }
 
+// GetServices returns a copy of the Services map for safe iteration by external callers.
+func (info *MetadataInfo) GetServices() map[string]*ServiceInfo {
+	info.mu.RLock()
+	defer info.mu.RUnlock()
+
+	cp := make(map[string]*ServiceInfo, len(info.Services))
+	for k, v := range info.Services {
+		cp[k] = v
+	}
+	return cp
+}
+
 // ServiceInfo the information of service
 type ServiceInfo struct {
 	Name     string            `json:"name,omitempty" hessian:"name"`

metadata/metadata.go

@@ -18,14 +18,19 @@
 // Package metadata collects and exposes information of all services for service discovery purpose.
 package metadata
 
+import (
+	"sync"
+)
+
 import (
 	"dubbo.apache.org/dubbo-go/v3/common"
 	"dubbo.apache.org/dubbo-go/v3/common/constant"
 	"dubbo.apache.org/dubbo-go/v3/metadata/info"
 )
 
 var (
-	registryMetadataInfo                 = make(map[string]*info.MetadataInfo)
+	registryMetadataInfo = make(map[string]*info.MetadataInfo)
+	registryMetadataLock sync.RWMutex
 	metadataService      MetadataService = &DefaultMetadataService{metadataMap: registryMetadataInfo}
 )
 
@@ -34,25 +39,33 @@ func GetMetadataService() MetadataService {
 }
 
 func GetMetadataInfo(registryId string) *info.MetadataInfo {
+	registryMetadataLock.RLock()
+	defer registryMetadataLock.RUnlock()
 	return registryMetadataInfo[registryId]
 }
 
 func AddService(registryId string, url *common.URL) {
+	registryMetadataLock.Lock()
 	if _, exist := registryMetadataInfo[registryId]; !exist {
 		registryMetadataInfo[registryId] = info.NewMetadataInfo(
 			url.GetParam(constant.ApplicationKey, ""),
 			url.GetParam(constant.ApplicationTagKey, ""),
 		)
 	}
+	registryMetadataLock.Unlock()
+
 	registryMetadataInfo[registryId].AddService(url)
 }
 
 func AddSubscribeURL(registryId string, url *common.URL) {
+	registryMetadataLock.Lock()
 	if _, exist := registryMetadataInfo[registryId]; !exist {
 		registryMetadataInfo[registryId] = info.NewMetadataInfo(
 			url.GetParam(constant.ApplicationKey, ""),
 			url.GetParam(constant.ApplicationTagKey, ""),
 		)
 	}
+	registryMetadataLock.Unlock()
+
 	registryMetadataInfo[registryId].AddSubscribeURL(url)
 }

metadata/metadata_service.go

@@ -257,7 +257,7 @@ func (mtsV2 *MetadataServiceV2) GetMetadataInfo(ctx context.Context, req *triple
 	return &tripleapi.MetadataInfoV2{
 		App:      metadataInfo.App,
 		Version:  metadataInfo.Revision,
-		Services: convertV2(metadataInfo.Services),
+		Services: convertV2(metadataInfo.GetServices()),
 	}, err
 }
 

metadata/report/nacos/report_test.go

@@ -171,7 +171,7 @@ func newNacosMetadataReport(f fields) *nacosMetadataReport {
 }
 
 func Test_nacosMetadataReport_GetAppMetadata(t *testing.T) {
-	mi := info.MetadataInfo{
+	mi := &info.MetadataInfo{
 		App: "GetAppMetadata",
 	}
 	data, _ := json.Marshal(mi)
@@ -198,7 +198,7 @@ func Test_nacosMetadataReport_GetAppMetadata(t *testing.T) {
 				application: "dubbo",
 				revision:    "revision",
 			},
-			want:    &mi,
+			want:    mi,
 			wantErr: false,
 		},
 	}

metadata/report_instance.go

@@ -18,6 +18,7 @@
 package metadata
 
 import (
+	"sync"
 	"time"
 )
 
@@ -38,7 +39,8 @@ import (
 )
 
 var (
-	instances = make(map[string]report.MetadataReport)
+	instances   = make(map[string]report.MetadataReport)
+	instancesMu sync.RWMutex
 )
 
 func addMetadataReport(registryId string, url *common.URL) error {
@@ -47,11 +49,19 @@ func addMetadataReport(registryId string, url *common.URL) error {
 		logger.Warnf("no metadata report factory of protocol %s found, please check if the metadata report factory is imported", url.Protocol)
 		return nil
 	}
+	instancesMu.Lock()
 	instances[registryId] = &DelegateMetadataReport{instance: fac.CreateMetadataReport(url)}
+	instancesMu.Unlock()
 	return nil
 }
 
 func GetMetadataReport() report.MetadataReport {
+	instancesMu.RLock()
+	defer instancesMu.RUnlock()
+	return getMetadataReportUnsafe()
+}
+
+func getMetadataReportUnsafe() report.MetadataReport {
 	for _, v := range instances {
 		return v
 	}
@@ -62,13 +72,17 @@ func GetMetadataReportByRegistry(registry string) report.MetadataReport {
 	if len(registry) == 0 {
 		registry = constant.DefaultKey
 	}
+	instancesMu.RLock()
+	defer instancesMu.RUnlock()
 	if r, ok := instances[registry]; ok {
 		return r
 	}
-	return GetMetadataReport()
+	return getMetadataReportUnsafe()
 }
 
 func GetMetadataReports() []report.MetadataReport {
+	instancesMu.RLock()
+	defer instancesMu.RUnlock()
 	reports := make([]report.MetadataReport, len(instances))
 	index := 0
 	for _, r := range instances {

registry/servicediscovery/service_instances_changed_listener_impl.go

@@ -134,7 +134,7 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e observer.Event) error
 				continue
 			}
 			instance.SetServiceMetadata(metadataInfo)
-			for _, service := range metadataInfo.Services {
+			for _, service := range metadataInfo.GetServices() {
 				matchKey := service.GetMatchKey()
 				if serviceToRevisionServices[matchKey] == nil {
 					serviceToRevisionServices[matchKey] = make(map[string]*info.ServiceInfo)
@@ -203,8 +203,11 @@ func toInstanceServiceURLs(instance registry.ServiceInstance, serviceInfo *info.
 
 // AddListenerAndNotify add notify listener and notify to listen service event
 func (lstn *ServiceInstancesChangedListenerImpl) AddListenerAndNotify(serviceKey string, notify registry.NotifyListener) {
+	lstn.mutex.Lock()
 	lstn.listeners[serviceKey] = notify
 	urls := lstn.serviceUrls[serviceKey]
+	lstn.mutex.Unlock()
+
 	for _, url := range urls {
 		notify.Notify(&registry.ServiceEvent{
 			Action:  remoting.EventTypeAdd,
@@ -215,7 +218,9 @@ func (lstn *ServiceInstancesChangedListenerImpl) AddListenerAndNotify(serviceKey
 
 // RemoveListener remove notify listener
 func (lstn *ServiceInstancesChangedListenerImpl) RemoveListener(serviceKey string) {
+	lstn.mutex.Lock()
 	delete(lstn.listeners, serviceKey)
+	lstn.mutex.Unlock()
 }
 
 // GetServiceNames return all listener service names