AsperforMias commented on code in PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#discussion_r3760893926
##########
registry/servicediscovery/service_discovery_registry.go:
##########
@@ -576,28 +582,185 @@ func (s *serviceDiscoveryRegistry)
getServiceListener(serviceNamesKey string) re
return s.serviceListeners[serviceNamesKey]
}
+// protocolServiceKeyOf builds the subscriber key SubscribeURL registers and
+// UnSubscribe removes: consumers default to the "tri" protocol, other
+// protocols need to be specified on the reference/consumer explicitly.
+func protocolServiceKeyOf(url *common.URL) string {
+ protocol := constant.TriProtocol
+ if url.Protocol != "" {
+ protocol = url.Protocol
+ }
+ return url.ServiceKey() + ":" + protocol
+}
+
+// loadLatestInstances pushes the current registry snapshot for every
subscribed
+// application into the listener. It runs outside s.lock: GetInstances and
+// OnEvent may perform external RPC / metadata-report calls.
+func (s *serviceDiscoveryRegistry) loadLatestInstances(listener
registry.ServiceInstancesChangedListener) {
+ for _, serviceNameTmp := range listener.GetServiceNames().Values() {
+ serviceName := serviceNameTmp.(string)
+ instances := s.serviceDiscovery.GetInstances(serviceName)
+ logger.Infof("[Registry][ServiceDiscovery] synchronized
instance notification on application %s subscription, instance list size %d",
serviceName, len(instances))
+ if err :=
listener.OnEvent(®istry.ServiceInstancesChangedEvent{
+ ServiceName: serviceName,
+ Instances: instances,
+ }); err != nil {
+ logger.Warnf("[Registry][ServiceDiscovery]
ServiceInstancesChangedListenerImpl handle error, err=%v", err)
+ }
+ }
+}
+
// subscribeAndNotify registers the notify callback and asynchronously wires
the
-// listener into the service discovery so the caller does not block on it.
+// listener into the service discovery so the caller does not block on it. A
+// failed AddListener is retried in the background with backoff; without the
+// retry a transient registry error would leave the consumer permanently stale
+// (issue #3624).
func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL,
serviceNamesKey, protocolServiceKey string,
listener registry.ServiceInstancesChangedListener, notify
registry.NotifyListener,
) {
listener.AddListenerAndNotify(protocolServiceKey, notify)
- event :=
metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
logger.Infof("[Registry][ServiceDiscovery] start subscribing to
registry for applications=%s with a new go routine", serviceNamesKey)
go func() {
- err := s.serviceDiscovery.AddListener(listener)
- event.Succ = err != nil
- event.End = time.Now()
- event.Attachment[constant.InterfaceKey] = url.Interface()
- metrics.Publish(event)
- metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err ==
nil))
- if err != nil {
+ if err := s.addInstanceListener(url, listener); err != nil {
logger.Errorf("[Registry][ServiceDiscovery] add
instance listener catch error, url=%s err=%s", url.String(), err.Error())
+ s.scheduleSubscribeRetry(serviceNamesKey,
&subscribeRetry{listener: listener, url: url})
}
}()
}
+// addInstanceListener installs the listener into the service discovery and
+// publishes the subscribe metrics.
+func (s *serviceDiscoveryRegistry) addInstanceListener(url *common.URL,
listener registry.ServiceInstancesChangedListener) error {
+ event :=
metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
+ err := s.serviceDiscovery.AddListener(listener)
+ event.Succ = err == nil
+ event.End = time.Now()
+ event.Attachment[constant.InterfaceKey] = url.Interface()
+ metrics.Publish(event)
+ metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err == nil))
+ return err
+}
+
+var (
+ // subscribeRetryInitialDelay is the first backoff delay before
retrying a
+ // failed AddListener call. Package-level so tests can shrink it.
+ subscribeRetryInitialDelay = time.Second
+ // subscribeRetryMaxDelay caps the backoff. The retry count itself is
+ // intentionally unlimited: a capped count would re-introduce the
+ // permanently stale consumer this mechanism fixes.
+ subscribeRetryMaxDelay = 30 * time.Second
+)
+
+// subscribeRetry is a pending AddListener retry for one serviceNamesKey.
+type subscribeRetry struct {
+ listener registry.ServiceInstancesChangedListener
+ url *common.URL
+ timer *time.Timer
+ attempts int
+}
+
+// scheduleSubscribeRetry arms the retry timer for serviceNamesKey after a
+// failed AddListener call. One pending retry per key: repeated failures share
+// the same timer instead of stacking new ones. Retries continue with capped
+// exponential backoff and jitter until the subscription is established, the
+// last subscriber unsubscribes, or the registry is destroyed.
+func (s *serviceDiscoveryRegistry) scheduleSubscribeRetry(serviceNamesKey
string, state *subscribeRetry) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ if s.destroyed {
+ return
+ }
+ if !listenerHasSubscribers(state.listener) {
+ // No subscriber left (e.g. unsubscribe raced with a failing
retry):
+ // do not arm a timer nobody waits for.
+ return
+ }
+ if _, ok := s.subscribeRetries[serviceNamesKey]; ok {
+ return
+ }
+ delay := subscribeRetryDelay(state.attempts)
+ state.attempts++
+ state.timer = time.AfterFunc(delay, func() {
+ s.retryAddListener(serviceNamesKey)
+ })
+ s.subscribeRetries[serviceNamesKey] = state
+ logger.Debugf("[Registry][ServiceDiscovery] instance listener for
applications=%s not established, retry in %s", serviceNamesKey, delay)
+}
+
+// cancelSubscribeRetry stops a pending AddListener retry, if any.
+func (s *serviceDiscoveryRegistry) cancelSubscribeRetry(serviceNamesKey
string) {
+ s.lock.Lock()
+ defer s.lock.Unlock()
+ s.cancelSubscribeRetryLocked(serviceNamesKey)
+}
+
+// cancelSubscribeRetryLocked stops a pending AddListener retry; caller must
+// hold s.lock.
+func (s *serviceDiscoveryRegistry) cancelSubscribeRetryLocked(serviceNamesKey
string) {
+ if state, ok := s.subscribeRetries[serviceNamesKey]; ok {
+ state.timer.Stop()
+ delete(s.subscribeRetries, serviceNamesKey)
+ }
+}
+
+// retryAddListener re-runs AddListener after the backoff delay. On success it
+// re-syncs the latest instance snapshot so instance changes missed while the
+// subscription was down are picked up instead of leaving the consumer on a
+// stale view.
+func (s *serviceDiscoveryRegistry) retryAddListener(serviceNamesKey string) {
+ s.lock.Lock()
+ state, ok := s.subscribeRetries[serviceNamesKey]
+ if !ok {
+ s.lock.Unlock()
+ return
+ }
+ delete(s.subscribeRetries, serviceNamesKey)
+ active := !s.destroyed &&
+ s.serviceListeners[serviceNamesKey] == state.listener &&
+ listenerHasSubscribers(state.listener)
+ s.lock.Unlock()
+
+ if !active {
+ // Registry destroyed, listener replaced, or no subscriber left:
+ // stop retrying so the loop cannot leak.
+ return
+ }
Review Comment:
done
##########
registry/servicediscovery/service_discovery_registry.go:
##########
@@ -576,28 +582,185 @@ func (s *serviceDiscoveryRegistry)
getServiceListener(serviceNamesKey string) re
return s.serviceListeners[serviceNamesKey]
}
+// protocolServiceKeyOf builds the subscriber key SubscribeURL registers and
+// UnSubscribe removes: consumers default to the "tri" protocol, other
+// protocols need to be specified on the reference/consumer explicitly.
+func protocolServiceKeyOf(url *common.URL) string {
+ protocol := constant.TriProtocol
+ if url.Protocol != "" {
+ protocol = url.Protocol
+ }
+ return url.ServiceKey() + ":" + protocol
+}
+
+// loadLatestInstances pushes the current registry snapshot for every
subscribed
+// application into the listener. It runs outside s.lock: GetInstances and
+// OnEvent may perform external RPC / metadata-report calls.
+func (s *serviceDiscoveryRegistry) loadLatestInstances(listener
registry.ServiceInstancesChangedListener) {
+ for _, serviceNameTmp := range listener.GetServiceNames().Values() {
+ serviceName := serviceNameTmp.(string)
+ instances := s.serviceDiscovery.GetInstances(serviceName)
+ logger.Infof("[Registry][ServiceDiscovery] synchronized
instance notification on application %s subscription, instance list size %d",
serviceName, len(instances))
+ if err :=
listener.OnEvent(®istry.ServiceInstancesChangedEvent{
+ ServiceName: serviceName,
+ Instances: instances,
+ }); err != nil {
+ logger.Warnf("[Registry][ServiceDiscovery]
ServiceInstancesChangedListenerImpl handle error, err=%v", err)
+ }
+ }
+}
+
// subscribeAndNotify registers the notify callback and asynchronously wires
the
-// listener into the service discovery so the caller does not block on it.
+// listener into the service discovery so the caller does not block on it. A
+// failed AddListener is retried in the background with backoff; without the
+// retry a transient registry error would leave the consumer permanently stale
+// (issue #3624).
func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL,
serviceNamesKey, protocolServiceKey string,
listener registry.ServiceInstancesChangedListener, notify
registry.NotifyListener,
) {
listener.AddListenerAndNotify(protocolServiceKey, notify)
- event :=
metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
logger.Infof("[Registry][ServiceDiscovery] start subscribing to
registry for applications=%s with a new go routine", serviceNamesKey)
go func() {
- err := s.serviceDiscovery.AddListener(listener)
- event.Succ = err != nil
- event.End = time.Now()
- event.Attachment[constant.InterfaceKey] = url.Interface()
- metrics.Publish(event)
- metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err ==
nil))
- if err != nil {
+ if err := s.addInstanceListener(url, listener); err != nil {
Review Comment:
done
--
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]
