This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/main by this push:
new f04c8d087 fix Panic caused by failure to connect to the peer when
obtaining remote metadata (#2651)
f04c8d087 is described below
commit f04c8d08753765f6028d2a84b01aa00e42175a04
Author: foghost <[email protected]>
AuthorDate: Mon Apr 29 11:39:17 2024 +0800
fix Panic caused by failure to connect to the peer when obtaining remote
metadata (#2651)
---
common/metadata_info.go | 32 ++++++++------
.../local/metadata_service_proxy_factory.go | 3 ++
.../servicediscovery/service_discovery_registry.go | 7 ++-
.../service_instances_changed_listener_impl.go | 51 ++++++++++------------
4 files changed, 50 insertions(+), 43 deletions(-)
diff --git a/common/metadata_info.go b/common/metadata_info.go
index aad5926b7..c9d83103e 100644
--- a/common/metadata_info.go
+++ b/common/metadata_info.go
@@ -33,20 +33,24 @@ import (
"dubbo.apache.org/dubbo-go/v3/common/constant"
)
-var IncludeKeys = gxset.NewSet(
- constant.ApplicationKey,
- constant.GroupKey,
- constant.TimestampKey,
- constant.SerializationKey,
- constant.ClusterKey,
- constant.LoadbalanceKey,
- constant.PathKey,
- constant.TimeoutKey,
- constant.TokenKey,
- constant.VersionKey,
- constant.WarmupKey,
- constant.WeightKey,
- constant.ReleaseKey)
+var (
+ IncludeKeys = gxset.NewSet(
+ constant.ApplicationKey,
+ constant.GroupKey,
+ constant.TimestampKey,
+ constant.SerializationKey,
+ constant.ClusterKey,
+ constant.LoadbalanceKey,
+ constant.PathKey,
+ constant.TimeoutKey,
+ constant.TokenKey,
+ constant.VersionKey,
+ constant.WarmupKey,
+ constant.WeightKey,
+ constant.ReleaseKey)
+
+ EmptyMetadataInfo = &MetadataInfo{}
+)
// MetadataInfo the metadata information of instance
type MetadataInfo struct {
diff --git a/metadata/service/local/metadata_service_proxy_factory.go
b/metadata/service/local/metadata_service_proxy_factory.go
index b91477805..9645374b3 100644
--- a/metadata/service/local/metadata_service_proxy_factory.go
+++ b/metadata/service/local/metadata_service_proxy_factory.go
@@ -67,6 +67,9 @@ func createProxy(ins registry.ServiceInstance)
service.MetadataService {
u := urls[0]
p := extension.GetProtocol(u.Protocol)
invoker := p.Refer(u)
+ if invoker == nil { // can't connect instance
+ return nil
+ }
return &MetadataServiceProxy{
Invoker: invoker,
}
diff --git a/registry/servicediscovery/service_discovery_registry.go
b/registry/servicediscovery/service_discovery_registry.go
index 6618544cd..70075db91 100644
--- a/registry/servicediscovery/service_discovery_registry.go
+++ b/registry/servicediscovery/service_discovery_registry.go
@@ -221,7 +221,10 @@ func (s *ServiceDiscoveryRegistry) Subscribe(url
*common.URL, notify registry.No
}
logger.Infof("Find initial mapping applications %q for service %s.",
services, url.ServiceKey())
// first notify
-
mappingListener.OnEvent(registry.NewServiceMappingChangedEvent(url.ServiceKey(),
services))
+ err =
mappingListener.OnEvent(registry.NewServiceMappingChangedEvent(url.ServiceKey(),
services))
+ if err != nil {
+ logger.Errorf("[ServiceDiscoveryRegistry]
ServiceInstancesChangedListenerImpl handle error:%v", err)
+ }
return nil
}
@@ -246,7 +249,7 @@ func (s *ServiceDiscoveryRegistry) SubscribeURL(url
*common.URL, notify registry
Instances: instances,
})
if err != nil {
- logger.Warnf("[ServiceDiscoveryRegistry]
ServiceInstancesChangedListenerImpl handle error:%v", err)
+ logger.Errorf("[ServiceDiscoveryRegistry]
ServiceInstancesChangedListenerImpl handle error:%v", err)
}
}
}
diff --git
a/registry/servicediscovery/service_instances_changed_listener_impl.go
b/registry/servicediscovery/service_instances_changed_listener_impl.go
index 5b5433049..ea21510b2 100644
--- a/registry/servicediscovery/service_instances_changed_listener_impl.go
+++ b/registry/servicediscovery/service_instances_changed_listener_impl.go
@@ -19,6 +19,7 @@ package servicediscovery
import (
"encoding/gob"
+ "errors"
"reflect"
"sync"
"time"
@@ -87,7 +88,6 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
if !ok {
return nil
}
- var err error
lstn.mutex.Lock()
defer lstn.mutex.Unlock()
@@ -119,15 +119,7 @@ func (lstn *ServiceInstancesChangedListenerImpl) OnEvent(e
observer.Event) error
revisionToInstances[revision] = append(subInstances,
instance)
metadataInfo := lstn.revisionToMetadata[revision]
if metadataInfo == nil {
- if val, ok := metaCache.Get(revision); ok {
- metadataInfo =
val.(*common.MetadataInfo)
- } else {
- metadataInfo, err =
GetMetadataInfo(lstn.app, instance, revision)
- if err != nil {
- return err
- }
- metaCache.Set(revision, metadataInfo)
- }
+ metadataInfo = GetMetadataInfo(lstn.app,
instance, revision)
}
instance.SetServiceMetadata(metadataInfo)
for _, service := range metadataInfo.Services {
@@ -225,44 +217,49 @@ func (lstn *ServiceInstancesChangedListenerImpl)
GetEventType() reflect.Type {
}
// GetMetadataInfo get metadata info when MetadataStorageTypePropertyName is
null
-func GetMetadataInfo(app string, instance registry.ServiceInstance, revision
string) (*common.MetadataInfo, error) {
+func GetMetadataInfo(app string, instance registry.ServiceInstance, revision
string) *common.MetadataInfo {
cacheOnce.Do(func() {
initCache(app)
})
if metadataInfo, ok := metaCache.Get(revision); ok {
- return metadataInfo.(*common.MetadataInfo), nil
+ return metadataInfo.(*common.MetadataInfo)
}
var metadataStorageType string
- var metadataInfo *common.MetadataInfo
+ metadataInfo := common.EmptyMetadataInfo
+ var err error
if instance.GetMetadata() == nil {
metadataStorageType = constant.DefaultMetadataStorageType
} else {
metadataStorageType =
instance.GetMetadata()[constant.MetadataStorageTypePropertyName]
}
if metadataStorageType == constant.RemoteMetadataStorageType {
- remoteMetadataServiceImpl, err :=
extension.GetRemoteMetadataService()
- if err != nil {
- return nil, err
- }
- metadataInfo, err =
remoteMetadataServiceImpl.GetMetadata(instance)
- if err != nil {
- return nil, err
+ remoteMetadataServiceImpl, remoteMetadataErr :=
extension.GetRemoteMetadataService()
+ if remoteMetadataErr == nil {
+ metadataInfo, err =
remoteMetadataServiceImpl.GetMetadata(instance)
+ } else {
+ err = remoteMetadataErr
}
} else {
- var err error
proxyFactory :=
extension.GetMetadataServiceProxyFactory(constant.DefaultKey)
metadataService := proxyFactory.GetProxy(instance)
- defer destroyInvoker(metadataService)
- metadataInfo, err = metadataService.GetMetadataInfo(revision)
- if err != nil {
- return nil, err
+ if metadataService != nil {
+ defer destroyInvoker(metadataService)
+ metadataInfo, err =
metadataService.GetMetadataInfo(revision)
+ } else {
+ err = errors.New("get remote metadata error please
check instance " + instance.GetHost() + " is alive")
}
}
- metaCache.Set(revision, metadataInfo)
+ if err != nil {
+ logger.Errorf("get metadata of %s failed, %v",
instance.GetHost(), err)
+ }
+
+ if metadataInfo != common.EmptyMetadataInfo {
+ metaCache.Set(revision, metadataInfo)
+ }
- return metadataInfo, nil
+ return metadataInfo
}
func destroyInvoker(metadataService service.MetadataService) {
