diff --git a/common/constant/default.go b/common/constant/default.go index 0680c4192d085237f2e1a460e668b14b26a9b32e..4165942a615e220f6384a898b07c04bafd39c3b0 100644 --- a/common/constant/default.go +++ b/common/constant/default.go @@ -37,6 +37,7 @@ const ( DEFAULT_RETRIES_INT = 2 DEFAULT_PROTOCOL = "dubbo" DEFAULT_REG_TIMEOUT = "10s" + DEFAULT_REG_TTL = "15m" DEFAULT_CLUSTER = "failover" DEFAULT_FAILBACK_TIMES = "3" DEFAULT_FAILBACK_TIMES_INT = 3 diff --git a/common/constant/key.go b/common/constant/key.go index c3fe0e79230d3cdf7f41186ac77ca4581d416331..c1d4c3adb7517262e7ffc4e3c2df18337c2d9046 100644 --- a/common/constant/key.go +++ b/common/constant/key.go @@ -99,6 +99,7 @@ const ( ROLE_KEY = "registry.role" REGISTRY_DEFAULT_KEY = "registry.default" REGISTRY_TIMEOUT_KEY = "registry.timeout" + REGISTRY_TTL_KEY = "registry.ttl" ) const ( diff --git a/config/base_config.go b/config/base_config.go index 0ba5bc7ef98cb30a13890b93a659c467adcbf73b..22a0832731daff6c9957d4913a3784c9b268b11f 100644 --- a/config/base_config.go +++ b/config/base_config.go @@ -29,11 +29,8 @@ import ( ) import ( - "github.com/apache/dubbo-go/common" "github.com/apache/dubbo-go/common/config" - "github.com/apache/dubbo-go/common/extension" "github.com/apache/dubbo-go/common/logger" - "github.com/apache/dubbo-go/config_center" ) type multiConfiger interface { @@ -52,7 +49,6 @@ type BaseConfig struct { // application config ApplicationConfig *ApplicationConfig `yaml:"application" json:"application,omitempty" property:"application"` - configCenterUrl *common.URL prefix string fatherConfig interface{} EventDispatcherType string `default:"direct" yaml:"event_dispatcher_type" json:"event_dispatcher_type,omitempty"` @@ -72,74 +68,6 @@ func (c *BaseConfig) GetRemoteConfig(name string) (config *RemoteConfig, ok bool return } -// startConfigCenter will start the config center. -// it will prepare the environment -func (c *BaseConfig) startConfigCenter() error { - url, err := common.NewURL(c.ConfigCenterConfig.Address, - common.WithProtocol(c.ConfigCenterConfig.Protocol), common.WithParams(c.ConfigCenterConfig.GetUrlMap())) - if err != nil { - return err - } - c.configCenterUrl = &url - if c.prepareEnvironment() != nil { - return perrors.WithMessagef(err, "start config center error!") - } - // c.fresh() - return err -} - -func (c *BaseConfig) prepareEnvironment() error { - factory := extension.GetConfigCenterFactory(c.ConfigCenterConfig.Protocol) - dynamicConfig, err := factory.GetDynamicConfiguration(c.configCenterUrl) - config.GetEnvInstance().SetDynamicConfiguration(dynamicConfig) - if err != nil { - logger.Errorf("Get dynamic configuration error , error message is %v", err) - return perrors.WithStack(err) - } - content, err := dynamicConfig.GetProperties(c.ConfigCenterConfig.ConfigFile, config_center.WithGroup(c.ConfigCenterConfig.Group)) - if err != nil { - logger.Errorf("Get config content in dynamic configuration error , error message is %v", err) - return perrors.WithStack(err) - } - var appGroup string - var appContent string - if providerConfig != nil && providerConfig.ApplicationConfig != nil && - reflect.ValueOf(c.fatherConfig).Elem().Type().Name() == "ProviderConfig" { - appGroup = providerConfig.ApplicationConfig.Name - } else if consumerConfig != nil && consumerConfig.ApplicationConfig != nil && - reflect.ValueOf(c.fatherConfig).Elem().Type().Name() == "ConsumerConfig" { - appGroup = consumerConfig.ApplicationConfig.Name - } - - if len(appGroup) != 0 { - configFile := c.ConfigCenterConfig.AppConfigFile - if len(configFile) == 0 { - configFile = c.ConfigCenterConfig.ConfigFile - } - appContent, err = dynamicConfig.GetProperties(configFile, config_center.WithGroup(appGroup)) - if err != nil { - return perrors.WithStack(err) - } - } - // global config file - mapContent, err := dynamicConfig.Parser().Parse(content) - if err != nil { - return perrors.WithStack(err) - } - config.GetEnvInstance().UpdateExternalConfigMap(mapContent) - - // appGroup config file - if len(appContent) != 0 { - appMapConent, err := dynamicConfig.Parser().Parse(appContent) - if err != nil { - return perrors.WithStack(err) - } - config.GetEnvInstance().UpdateAppExternalConfigMap(appMapConent) - } - - return nil -} - func getKeyPrefix(val reflect.Value) []string { var ( prefix string diff --git a/config/base_config_test.go b/config/base_config_test.go index 6db6a8dcb84de3fdefe94cce87338b9efe28246c..849a9c4586c0c8cd2d74e3dd1011aaab466f0e93 100644 --- a/config/base_config_test.go +++ b/config/base_config_test.go @@ -28,8 +28,6 @@ import ( import ( "github.com/apache/dubbo-go/common/config" - "github.com/apache/dubbo-go/common/extension" - "github.com/apache/dubbo-go/config_center" _ "github.com/apache/dubbo-go/config_center/apollo" ) @@ -282,23 +280,6 @@ func TestRefreshProvider(t *testing.T) { assert.Equal(t, "20001", father.Protocols["jsonrpc1"].Port) } -func TestStartConfigCenter(t *testing.T) { - extension.SetConfigCenterFactory("mock", func() config_center.DynamicConfigurationFactory { - return &config_center.MockDynamicConfigurationFactory{} - }) - c := &BaseConfig{ConfigCenterConfig: &ConfigCenterConfig{ - Protocol: "mock", - Address: "172.0.0.1", - Group: "dubbo", - ConfigFile: "mockDubbo.properties", - }} - err := c.startConfigCenter() - assert.NoError(t, err) - b, v := config.GetEnvInstance().Configuration().Back().Value.(*config.InmemoryConfiguration).GetProperty("dubbo.application.organization") - assert.True(t, b) - assert.Equal(t, "ikurento.com", v) -} - func TestInitializeStruct(t *testing.T) { testConsumerConfig := &ConsumerConfig{} tp := reflect.TypeOf(ConsumerConfig{}) diff --git a/config/config_center_config.go b/config/config_center_config.go index c9133dc26df0b05e3bb61df0f612d0e2914e98bb..0fc4007940d9b1ac2456c9b2d379493bb5d8edb0 100644 --- a/config/config_center_config.go +++ b/config/config_center_config.go @@ -20,6 +20,7 @@ package config import ( "context" "net/url" + "reflect" "time" ) @@ -28,7 +29,13 @@ import ( ) import ( + "github.com/apache/dubbo-go/common" + "github.com/apache/dubbo-go/common/config" "github.com/apache/dubbo-go/common/constant" + "github.com/apache/dubbo-go/common/extension" + "github.com/apache/dubbo-go/common/logger" + "github.com/apache/dubbo-go/config_center" + perrors "github.com/pkg/errors" ) // ConfigCenterConfig is configuration for config center @@ -52,6 +59,7 @@ type ConfigCenterConfig struct { AppConfigFile string `default:"dubbo.properties" yaml:"app_config_file" json:"app_config_file,omitempty"` AppId string `default:"dubbo" yaml:"app_id" json:"app_id,omitempty"` TimeoutStr string `yaml:"timeout" json:"timeout,omitempty"` + RemoteRef string `required:"false" yaml:"remote_ref" json:"remote_ref,omitempty"` timeout time.Duration } @@ -77,3 +85,94 @@ func (c *ConfigCenterConfig) GetUrlMap() url.Values { urlMap.Set(constant.CONFIG_LOG_DIR_KEY, c.LogDir) return urlMap } + +type configCenter struct { +} + +// toURL will compatible with baseConfig.ConfigCenterConfig.Address and baseConfig.ConfigCenterConfig.RemoteRef before 1.6.0 +// After 1.6.0 will not compatible, only baseConfig.ConfigCenterConfig.RemoteRef +func (b *configCenter) toURL(baseConfig BaseConfig) (common.URL, error) { + if len(baseConfig.ConfigCenterConfig.Address) > 0 { + return common.NewURL(baseConfig.ConfigCenterConfig.Address, + common.WithProtocol(baseConfig.ConfigCenterConfig.Protocol), common.WithParams(baseConfig.ConfigCenterConfig.GetUrlMap())) + } + + remoteRef := baseConfig.ConfigCenterConfig.RemoteRef + rc, ok := baseConfig.GetRemoteConfig(remoteRef) + + if !ok { + return common.URL{}, perrors.New("Could not find out the remote ref config, name: " + remoteRef) + } + + newURL, err := rc.toURL() + if err == nil { + newURL.SetParams(baseConfig.ConfigCenterConfig.GetUrlMap()) + } + return newURL, err +} + +// startConfigCenter will start the config center. +// it will prepare the environment +func (b *configCenter) startConfigCenter(baseConfig BaseConfig) error { + url, err := b.toURL(baseConfig) + if err != nil { + return err + } + if err = b.prepareEnvironment(baseConfig, &url); err != nil { + return perrors.WithMessagef(err, "start config center error!") + } + // c.fresh() + return nil +} + +func (b *configCenter) prepareEnvironment(baseConfig BaseConfig, configCenterUrl *common.URL) error { + factory := extension.GetConfigCenterFactory(configCenterUrl.Protocol) + dynamicConfig, err := factory.GetDynamicConfiguration(configCenterUrl) + if err != nil { + logger.Errorf("Get dynamic configuration error , error message is %v", err) + return perrors.WithStack(err) + } + config.GetEnvInstance().SetDynamicConfiguration(dynamicConfig) + content, err := dynamicConfig.GetProperties(baseConfig.ConfigCenterConfig.ConfigFile, config_center.WithGroup(baseConfig.ConfigCenterConfig.Group)) + if err != nil { + logger.Errorf("Get config content in dynamic configuration error , error message is %v", err) + return perrors.WithStack(err) + } + var appGroup string + var appContent string + if providerConfig != nil && providerConfig.ApplicationConfig != nil && + reflect.ValueOf(baseConfig.fatherConfig).Elem().Type().Name() == "ProviderConfig" { + appGroup = providerConfig.ApplicationConfig.Name + } else if consumerConfig != nil && consumerConfig.ApplicationConfig != nil && + reflect.ValueOf(baseConfig.fatherConfig).Elem().Type().Name() == "ConsumerConfig" { + appGroup = consumerConfig.ApplicationConfig.Name + } + + if len(appGroup) != 0 { + configFile := baseConfig.ConfigCenterConfig.AppConfigFile + if len(configFile) == 0 { + configFile = baseConfig.ConfigCenterConfig.ConfigFile + } + appContent, err = dynamicConfig.GetProperties(configFile, config_center.WithGroup(appGroup)) + if err != nil { + return perrors.WithStack(err) + } + } + // global config file + mapContent, err := dynamicConfig.Parser().Parse(content) + if err != nil { + return perrors.WithStack(err) + } + config.GetEnvInstance().UpdateExternalConfigMap(mapContent) + + // appGroup config file + if len(appContent) != 0 { + appMapConent, err := dynamicConfig.Parser().Parse(appContent) + if err != nil { + return perrors.WithStack(err) + } + config.GetEnvInstance().UpdateAppExternalConfigMap(appMapConent) + } + + return nil +} diff --git a/config/config_center_config_test.go b/config/config_center_config_test.go new file mode 100644 index 0000000000000000000000000000000000000000..2299167bb62a019f6854ea48d61442f5bde7e646 --- /dev/null +++ b/config/config_center_config_test.go @@ -0,0 +1,93 @@ +/* + * 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 config + +import ( + "testing" +) + +import ( + "github.com/stretchr/testify/assert" +) + +import ( + "github.com/apache/dubbo-go/common/config" + "github.com/apache/dubbo-go/common/extension" + "github.com/apache/dubbo-go/config_center" +) + +func TestStartConfigCenter(t *testing.T) { + extension.SetConfigCenterFactory("mock", func() config_center.DynamicConfigurationFactory { + return &config_center.MockDynamicConfigurationFactory{} + }) + baseConfig := &BaseConfig{ConfigCenterConfig: &ConfigCenterConfig{ + Protocol: "mock", + Address: "172.0.0.1", + Group: "dubbo", + ConfigFile: "mockDubbo.properties", + }} + + c := &configCenter{} + err := c.startConfigCenter(*baseConfig) + assert.NoError(t, err) + b, v := config.GetEnvInstance().Configuration().Back().Value.(*config.InmemoryConfiguration).GetProperty("dubbo.application.organization") + assert.True(t, b) + assert.Equal(t, "ikurento.com", v) +} + +func TestStartConfigCenterWithRemoteRef(t *testing.T) { + extension.SetConfigCenterFactory("mock", func() config_center.DynamicConfigurationFactory { + return &config_center.MockDynamicConfigurationFactory{} + }) + m := make(map[string]*RemoteConfig) + m["mock"] = &RemoteConfig{Protocol: "mock", Address: "172.0.0.1"} + baseConfig := &BaseConfig{ + Remotes: m, + ConfigCenterConfig: &ConfigCenterConfig{ + Group: "dubbo", + RemoteRef: "mock", + ConfigFile: "mockDubbo.properties", + }} + + c := &configCenter{} + err := c.startConfigCenter(*baseConfig) + assert.NoError(t, err) + b, v := config.GetEnvInstance().Configuration().Back().Value.(*config.InmemoryConfiguration).GetProperty("dubbo.application.organization") + assert.True(t, b) + assert.Equal(t, "ikurento.com", v) +} + +func TestStartConfigCenterWithRemoteRefError(t *testing.T) { + extension.SetConfigCenterFactory("mock", func() config_center.DynamicConfigurationFactory { + return &config_center.MockDynamicConfigurationFactory{} + }) + m := make(map[string]*RemoteConfig) + m["mock"] = &RemoteConfig{Address: "172.0.0.1"} + baseConfig := &BaseConfig{ + Remotes: m, + ConfigCenterConfig: &ConfigCenterConfig{ + Protocol: "mock", + Group: "dubbo", + RemoteRef: "mock", + ConfigFile: "mockDubbo.properties", + }} + + c := &configCenter{} + err := c.startConfigCenter(*baseConfig) + assert.Error(t, err) +} diff --git a/config/consumer_config.go b/config/consumer_config.go index 48f29f0e70028a7c057ee3831b45afa72446f3d0..177531209225b7a4b25751352d9777c0d4ee260c 100644 --- a/config/consumer_config.go +++ b/config/consumer_config.go @@ -41,7 +41,8 @@ import ( // ConsumerConfig is Consumer default configuration type ConsumerConfig struct { BaseConfig `yaml:",inline"` - Filter string `yaml:"filter" json:"filter,omitempty" property:"filter"` + configCenter + Filter string `yaml:"filter" json:"filter,omitempty" property:"filter"` // client Connect_Timeout string `default:"100ms" yaml:"connect_timeout" json:"connect_timeout,omitempty" property:"connect_timeout"` ConnectTimeout time.Duration @@ -125,13 +126,6 @@ func ConsumerInit(confConFile string) error { func configCenterRefreshConsumer() error { //fresh it var err error - if consumerConfig.ConfigCenterConfig != nil { - consumerConfig.SetFatherConfig(consumerConfig) - if err = consumerConfig.startConfigCenter(); err != nil { - return perrors.Errorf("start config center error , error message is {%v}", perrors.WithStack(err)) - } - consumerConfig.fresh() - } if consumerConfig.Request_Timeout != "" { if consumerConfig.RequestTimeout, err = time.ParseDuration(consumerConfig.Request_Timeout); err != nil { return perrors.WithMessagef(err, "time.ParseDuration(Request_Timeout{%#v})", consumerConfig.Request_Timeout) @@ -142,5 +136,12 @@ func configCenterRefreshConsumer() error { return perrors.WithMessagef(err, "time.ParseDuration(Connect_Timeout{%#v})", consumerConfig.Connect_Timeout) } } + if consumerConfig.ConfigCenterConfig != nil { + consumerConfig.SetFatherConfig(consumerConfig) + if err = consumerConfig.startConfigCenter((*consumerConfig).BaseConfig); err != nil { + return perrors.Errorf("start config center error , error message is {%v}", perrors.WithStack(err)) + } + consumerConfig.fresh() + } return nil } diff --git a/config/provider_config.go b/config/provider_config.go index 7cd3c1e98bfb8c35abb2b414b782ec709d0a8d0d..c710e48dc233a62837b31a89828e9c612eaff093 100644 --- a/config/provider_config.go +++ b/config/provider_config.go @@ -37,7 +37,8 @@ import ( // ProviderConfig is the default configuration of service provider type ProviderConfig struct { - BaseConfig `yaml:",inline"` + BaseConfig `yaml:",inline"` + configCenter Filter string `yaml:"filter" json:"filter,omitempty" property:"filter"` ProxyFactory string `yaml:"proxy_factory" default:"default" json:"proxy_factory,omitempty" property:"proxy_factory"` Services map[string]*ServiceConfig `yaml:"services" json:"services,omitempty" property:"services"` @@ -101,7 +102,7 @@ func configCenterRefreshProvider() error { // fresh it if providerConfig.ConfigCenterConfig != nil { providerConfig.fatherConfig = providerConfig - if err := providerConfig.startConfigCenter(); err != nil { + if err := providerConfig.startConfigCenter((*providerConfig).BaseConfig); err != nil { return perrors.Errorf("start config center error , error message is {%v}", perrors.WithStack(err)) } providerConfig.fresh() diff --git a/config/registry_config.go b/config/registry_config.go index ef527c827e9dac4cd2762f579d30254e9e51150f..703606b836906a8dbe9964e74da0edbe76991f48 100644 --- a/config/registry_config.go +++ b/config/registry_config.go @@ -39,6 +39,7 @@ type RegistryConfig struct { // I changed "type" to "protocol" ,the same as "protocol" field in java class RegistryConfig TimeoutStr string `yaml:"timeout" default:"5s" json:"timeout,omitempty" property:"timeout"` // unit: second Group string `yaml:"group" json:"group,omitempty" property:"group"` + TTL string `yaml:"ttl" default:"10m" json:"ttl,omitempty" property:"ttl"` // unit: minute // for registry Address string `yaml:"address" json:"address,omitempty" property:"address"` Username string `yaml:"username" json:"username,omitempty" property:"username"` @@ -118,6 +119,7 @@ func (c *RegistryConfig) getUrlMap(roleType common.RoleType) url.Values { urlMap.Set(constant.ROLE_KEY, strconv.Itoa(int(roleType))) urlMap.Set(constant.REGISTRY_KEY, c.Protocol) urlMap.Set(constant.REGISTRY_TIMEOUT_KEY, c.TimeoutStr) + urlMap.Set(constant.REGISTRY_TTL_KEY, c.TTL) for k, v := range c.Params { urlMap.Set(k, v) } diff --git a/config/remote_config.go b/config/remote_config.go index 5e0330c571715d99e63688ee944c61f8e48117bb..55380dd5a05b47b5b4677b32daf73b37376673d0 100644 --- a/config/remote_config.go +++ b/config/remote_config.go @@ -22,6 +22,11 @@ import ( ) import ( + perrors "github.com/pkg/errors" +) + +import ( + "github.com/apache/dubbo-go/common" "github.com/apache/dubbo-go/common/logger" ) @@ -30,6 +35,7 @@ import ( // so that other module, like config center, registry could reuse the config // but now, only metadata report, metadata service, service discovery use this structure type RemoteConfig struct { + Protocol string `yaml:"protocol" json:"protocol,omitempty"` Address string `yaml:"address" json:"address,omitempty"` TimeoutStr string `default:"5s" yaml:"timeout" json:"timeout,omitempty"` Username string `yaml:"username" json:"username,omitempty" property:"username"` @@ -56,3 +62,15 @@ func (rc *RemoteConfig) GetParam(key string, def string) string { } return param } + +func (rc *RemoteConfig) toURL() (common.URL, error) { + if len(rc.Protocol) == 0 { + return common.URL{}, perrors.Errorf("Must provide protocol in RemoteConfig.") + } + return common.NewURL(rc.Address, + common.WithUsername(rc.Username), + common.WithPassword(rc.Password), + common.WithLocation(rc.Address), + common.WithProtocol(rc.Protocol), + ) +} diff --git a/config_center/nacos/client.go b/config_center/nacos/client.go index 6fe5c4d7df28a7693c732543140ed74f959dc77e..9a09b713fad11afdc06310a2d0072454342ccb0b 100644 --- a/config_center/nacos/client.go +++ b/config_center/nacos/client.go @@ -167,7 +167,6 @@ func initNacosConfigClient(nacosAddrs []string, timeout time.Duration, url commo "serverConfigs": svrConfList, "clientConfig": nacosconst.ClientConfig{ TimeoutMs: uint64(int32(timeout / time.Millisecond)), - ListenInterval: uint64(int32(timeout / time.Millisecond)), NotLoadCacheAtStart: true, LogDir: url.GetParam(constant.NACOS_LOG_DIR_KEY, logDir), CacheDir: url.GetParam(constant.NACOS_CACHE_DIR_KEY, ""), diff --git a/go.mod b/go.mod index 7a472daef7481be390c3debfa00b1d1cd5207c47..c19627378261221b8a9565730ed62f7cc367e5e2 100644 --- a/go.mod +++ b/go.mod @@ -42,7 +42,7 @@ require ( github.com/mitchellh/hashstructure v1.0.0 // indirect github.com/mitchellh/mapstructure v1.2.3 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd - github.com/nacos-group/nacos-sdk-go v0.3.3-0.20200617023039-50c7537d6a5f + github.com/nacos-group/nacos-sdk-go v1.0.0 github.com/opentracing/opentracing-go v1.1.0 github.com/pierrec/lz4 v2.2.6+incompatible // indirect github.com/pkg/errors v0.9.1 diff --git a/go.sum b/go.sum index 424a11087bb9cad0c62488e8e3e96d64feec6d7d..aa6ecc86e25276eae2747aa5291f718713d164bd 100644 --- a/go.sum +++ b/go.sum @@ -516,8 +516,8 @@ github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3Rllmb github.com/munnerz/goautoneg v0.0.0-20120707110453-a547fc61f48d/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f/go.mod h1:ZdcZmHo+o7JKHSa8/e818NopupXU1YMK5fe1lsApnBw= -github.com/nacos-group/nacos-sdk-go v0.3.3-0.20200617023039-50c7537d6a5f h1:gid5/0AkHvINWK69Fgbidb3BVIXqlf1YEm7wO0NVPsw= -github.com/nacos-group/nacos-sdk-go v0.3.3-0.20200617023039-50c7537d6a5f/go.mod h1:fti1GlX/EB6RDKvzK/P7Vuibqj0JMPJHQwrcTU1tLXk= +github.com/nacos-group/nacos-sdk-go v1.0.0 h1:CufUF7DZca2ZzIrJtMMCDih1sA58BWCglArLMCZArUc= +github.com/nacos-group/nacos-sdk-go v1.0.0/go.mod h1:hlAPn3UdzlxIlSILAyOXKxjFSvDJ9oLzTJ9hLAK1KzA= github.com/nicolai86/scaleway-sdk v1.10.2-0.20180628010248-798f60e20bb2 h1:BQ1HW7hr4IVovMwWg0E0PYcyW8CzqDcVmaew9cujU4s= github.com/nicolai86/scaleway-sdk v1.10.2-0.20180628010248-798f60e20bb2/go.mod h1:TLb2Sg7HQcgGdloNxkrmtgDNR9uVYF3lfdFIN4Ro6Sk= github.com/oklog/run v1.0.0 h1:Ru7dDtJNOyC66gQ5dQmaCa0qIsAUFY3sFpK1Xk8igrw= diff --git a/registry/nacos/registry.go b/registry/nacos/registry.go index 51d3e2f56abac8e4ab8b966870f1ff5bb79c4171..411090820c7682ab9c3b5576ea8ad5207c2c899f 100644 --- a/registry/nacos/registry.go +++ b/registry/nacos/registry.go @@ -59,6 +59,7 @@ func init() { type nacosRegistry struct { *common.URL namingClient naming_client.INamingClient + registryUrls []common.URL } func getCategory(url common.URL) string { @@ -128,6 +129,36 @@ func (nr *nacosRegistry) Register(url common.URL) error { if !isRegistry { return perrors.New("registry [" + serviceName + "] to nacos failed") } + nr.registryUrls = append(nr.registryUrls, url) + return nil +} + +func createDeregisterParam(url common.URL, serviceName string) vo.DeregisterInstanceParam { + if len(url.Ip) == 0 { + url.Ip = localIP + } + if len(url.Port) == 0 || url.Port == "0" { + url.Port = "80" + } + port, _ := strconv.Atoi(url.Port) + return vo.DeregisterInstanceParam{ + Ip: url.Ip, + Port: uint64(port), + ServiceName: serviceName, + Ephemeral: true, + } +} + +func (nr *nacosRegistry) DeRegister(url common.URL) error { + serviceName := getServiceName(url) + param := createDeregisterParam(url, serviceName) + isDeRegistry, err := nr.namingClient.DeregisterInstance(param) + if err != nil { + return err + } + if !isDeRegistry { + return perrors.New("DeRegistry [" + serviceName + "] to nacos failed") + } return nil } @@ -193,6 +224,13 @@ func (nr *nacosRegistry) IsAvailable() bool { // nolint func (nr *nacosRegistry) Destroy() { + for _, url := range nr.registryUrls { + err := nr.DeRegister(url) + logger.Infof("DeRegister Nacos URL:%+v", url) + if err != nil { + logger.Errorf("Deregister URL:%+v err:%v", url, err.Error()) + } + } return } @@ -209,6 +247,7 @@ func newNacosRegistry(url *common.URL) (registry.Registry, error) { registry := &nacosRegistry{ URL: url, namingClient: client, + registryUrls: []common.URL{}, } return registry, nil } diff --git a/registry/nacos/service_discovery.go b/registry/nacos/service_discovery.go index 63d92d70fd5e1a00f0ce1ca95b1926fb9c36c84b..0e5ad8e6990856aeb0dfdde72f9c3f7fdae3e985 100644 --- a/registry/nacos/service_discovery.go +++ b/registry/nacos/service_discovery.go @@ -60,11 +60,20 @@ type nacosServiceDiscovery struct { // namingClient is the Nacos' client namingClient naming_client.INamingClient + // cache registry instances + registryInstances []registry.ServiceInstance } // Destroy will close the service discovery. // Actually, it only marks the naming client as null and then return func (n *nacosServiceDiscovery) Destroy() error { + for _, inst := range n.registryInstances { + err := n.Unregister(inst) + logger.Infof("Unregister nacos instance:%+v", inst) + if err != nil { + logger.Errorf("Unregister nacos instance:%+v, err:%+v", inst, err) + } + } n.namingClient = nil return nil } @@ -76,6 +85,7 @@ func (n *nacosServiceDiscovery) Register(instance registry.ServiceInstance) erro if err != nil || !ok { return perrors.WithMessage(err, "Could not register the instance. "+instance.GetServiceName()) } + n.registryInstances = append(n.registryInstances, instance) return nil } @@ -118,8 +128,8 @@ func (n *nacosServiceDiscovery) GetServices() *gxset.HashSet { return res } - for _, e := range services { - res.Add(e.Name) + for _, e := range services.Doms { + res.Add(e) } return res } @@ -334,8 +344,9 @@ func newNacosServiceDiscovery(name string) (registry.ServiceDiscovery, error) { descriptor := fmt.Sprintf("nacos-service-discovery[%s]", remoteConfig.Address) return &nacosServiceDiscovery{ - group: group, - namingClient: client, - descriptor: descriptor, + group: group, + namingClient: client, + descriptor: descriptor, + registryInstances: []registry.ServiceInstance{}, }, nil } diff --git a/registry/zookeeper/registry.go b/registry/zookeeper/registry.go index 8f2ac1023b8ad34938b9996b480e3bbc4adbaaea..e8ee51beb70b5a08ec60b213c5342ef52972c59f 100644 --- a/registry/zookeeper/registry.go +++ b/registry/zookeeper/registry.go @@ -243,6 +243,8 @@ func (r *zkRegistry) getListener(conf *common.URL) (*RegistryConfigurationListen var zkListener *RegistryConfigurationListener dataListener := r.dataListener + ttl := r.GetParam(constant.REGISTRY_TTL_KEY, constant.DEFAULT_REG_TTL) + conf.SetParam(constant.REGISTRY_TTL_KEY, ttl) dataListener.mutex.Lock() defer dataListener.mutex.Unlock() if r.dataListener.subscribed[conf.ServiceKey()] != nil { diff --git a/remoting/kubernetes/listener_test.go b/remoting/kubernetes/listener_test.go index 1f398485b2f16defddf44ce1a08a7ecfd9760dd1..0b05b6e6e09501dfd25164e865a3eb6eef91dc9f 100644 --- a/remoting/kubernetes/listener_test.go +++ b/remoting/kubernetes/listener_test.go @@ -19,6 +19,7 @@ package kubernetes import ( "testing" + "time" ) import ( @@ -87,6 +88,7 @@ func TestListener(t *testing.T) { listener := NewEventListener(c) dataListener := &mockDataListener{client: c, changedData: changedData, rc: make(chan remoting.Event)} listener.ListenServiceEvent("/dubbo", dataListener) + time.Sleep(1e9) for _, tc := range tests { diff --git a/remoting/zookeeper/listener.go b/remoting/zookeeper/listener.go index 9a4874db24696d90e4fcc7d9d987f5888f1be599..4f50c18ab61ba6faf373dfd0f831c14ae7ab6d5d 100644 --- a/remoting/zookeeper/listener.go +++ b/remoting/zookeeper/listener.go @@ -37,6 +37,10 @@ import ( "github.com/apache/dubbo-go/remoting" ) +var ( + defaultTTL = 15 * time.Minute +) + // nolint type ZkEventListener struct { client *ZookeeperClient @@ -197,10 +201,20 @@ func (l *ZkEventListener) listenDirEvent(conf *common.URL, zkPath string, listen var ( failTimes int + ttl time.Duration event chan struct{} zkEvent zk.Event ) event = make(chan struct{}, 4) + ttl = defaultTTL + if conf != nil { + timeout, err := time.ParseDuration(conf.GetParam(constant.REGISTRY_TTL_KEY, constant.DEFAULT_REG_TTL)) + if err == nil { + ttl = timeout + } else { + logger.Warnf("wrong configuration for registry ttl, error:=%+v, using default value %v instead", err, defaultTTL) + } + } defer close(event) for { // get current children for a zkPath @@ -302,18 +316,29 @@ func (l *ZkEventListener) listenDirEvent(conf *common.URL, zkPath string, listen }(dubboPath, listener) } } - select { - case zkEvent = <-childEventCh: - logger.Warnf("get a zookeeper zkEvent{type:%s, server:%s, path:%s, state:%d-%s, err:%s}", - zkEvent.Type.String(), zkEvent.Server, zkEvent.Path, zkEvent.State, StateToString(zkEvent.State), zkEvent.Err) - if zkEvent.Type != zk.EventNodeChildrenChanged { - continue + // Periodically update provider information + ticker := time.NewTicker(ttl) + WATCH: + for { + select { + case <-ticker.C: + l.handleZkNodeEvent(zkEvent.Path, children, listener) + case zkEvent = <-childEventCh: + logger.Warnf("get a zookeeper zkEvent{type:%s, server:%s, path:%s, state:%d-%s, err:%s}", + zkEvent.Type.String(), zkEvent.Server, zkEvent.Path, zkEvent.State, StateToString(zkEvent.State), zkEvent.Err) + ticker.Stop() + if zkEvent.Type != zk.EventNodeChildrenChanged { + break WATCH + } + l.handleZkNodeEvent(zkEvent.Path, children, listener) + break WATCH + case <-l.client.Done(): + logger.Warnf("client.done(), listen(path{%s}) goroutine exit now...", zkPath) + ticker.Stop() + return } - l.handleZkNodeEvent(zkEvent.Path, children, listener) - case <-l.client.Done(): - logger.Warnf("client.done(), listen(path{%s}) goroutine exit now...", zkPath) - return } + } }