directory.go

/*
 * 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 directory

import (
	"fmt"
	"net/url"
	"os"
	"reflect"
	"sync"
	"time"
)

import (
	"github.com/dubbogo/gost/log/logger"

	perrors "github.com/pkg/errors"
)

import (
	"dubbo.apache.org/dubbo-go/v3/cluster/directory"
	"dubbo.apache.org/dubbo-go/v3/cluster/directory/base"
	"dubbo.apache.org/dubbo-go/v3/cluster/directory/static"
	"dubbo.apache.org/dubbo-go/v3/cluster/router/chain"
	"dubbo.apache.org/dubbo-go/v3/common"
	commonConfig "dubbo.apache.org/dubbo-go/v3/common/config"
	"dubbo.apache.org/dubbo-go/v3/common/constant"
	"dubbo.apache.org/dubbo-go/v3/common/extension"
	"dubbo.apache.org/dubbo-go/v3/config_center"
	_ "dubbo.apache.org/dubbo-go/v3/config_center/configurator"
	"dubbo.apache.org/dubbo-go/v3/global"
	"dubbo.apache.org/dubbo-go/v3/graceful_shutdown"
	"dubbo.apache.org/dubbo-go/v3/metrics"
	metricsRegistry "dubbo.apache.org/dubbo-go/v3/metrics/registry"
	protocolbase "dubbo.apache.org/dubbo-go/v3/protocol/base"
	"dubbo.apache.org/dubbo-go/v3/protocol/protocolwrapper"
	"dubbo.apache.org/dubbo-go/v3/registry"
	"dubbo.apache.org/dubbo-go/v3/remoting"
)

func init() {
	extension.SetDefaultRegistryDirectory(NewRegistryDirectory)
	extension.SetDirectory(constant.RegistryProtocol, NewRegistryDirectory)
	extension.SetDirectory(constant.ServiceRegistryProtocol, NewServiceDiscoveryRegistryDirectory)
}

// RegistryDirectory implementation of Directory:
// Invoker list returned from this Directory's list method have been filtered by Routers
type RegistryDirectory struct {
	*base.Directory
	cacheInvokers                  []protocolbase.Invoker
	invokersLock                   sync.RWMutex
	serviceType                    string
	registry                       registry.Registry
	cacheInvokersMap               *sync.Map // use sync.map
	consumerURL                    *common.URL
	cacheOriginUrl                 *common.URL
	cachedProviderEvents           []*registry.ServiceEvent
	configurators                  []config_center.Configurator
	configuratorsLock              sync.RWMutex
	consumerConfigurationListener  *consumerConfigurationListener
	referenceConfigurationListener *referenceConfigurationListener
	registerLock                   sync.Mutex // this lock if for register
	subscribedUrlLock              sync.RWMutex
	SubscribedUrl                  *common.URL
	RegisteredUrl                  *common.URL
	closingTombstones              *sync.Map // map[string]closingTombstone
	closingTombstoneTTL            time.Duration
}

type closingTombstone struct {
	InstanceKey string
	ServiceKey  string
	Address     string
	// Timestamp is the export timestamp of the closing instance's URL. A re-add
	// carrying a different timestamp is a genuine restart, not a stale
	// pre-shutdown registry snapshot.
	Timestamp string
	Source    string
	ExpireAt  time.Time
}

var defaultClosingTombstoneTTL = func() time.Duration {
	if duration, err := time.ParseDuration(global.DefaultShutdownConfig().ClosingInvokerExpireTime); err == nil && duration > 0 {
		return duration
	}
	return 30 * time.Second
}()

func ensureRegistriesAttribute(url *common.URL) {
	if registries, ok := registriesFromAttribute(url); ok {
		url.SetAttribute(constant.RegistriesConfigKey, registries)
		return
	}
	if registries, ok := registriesFromAttribute(url.SubURL); ok {
		url.SetAttribute(constant.RegistriesConfigKey, registries)
		return
	}
	url.SetAttribute(constant.RegistriesConfigKey, map[string]*global.RegistryConfig{
		constant.DefaultKey: global.DefaultRegistryConfig(),
	})
}

func registriesFromAttribute(url *common.URL) (map[string]*global.RegistryConfig, bool) {
	if url == nil {
		return nil, false
	}

	registriesRaw, ok := url.GetAttribute(constant.RegistriesConfigKey)
	if !ok {
		return nil, false
	}

	switch registries := registriesRaw.(type) {
	case map[string]*global.RegistryConfig:
		if registries != nil {
			return registries, true
		}
	case map[string]global.RegistryConfig:
		if registries != nil {
			converted := make(map[string]*global.RegistryConfig, len(registries))
			for key, registryConfig := range registries {
				registryConfigCopy := registryConfig
				converted[key] = &registryConfigCopy
			}
			return converted, true
		}
	}
	return nil, false
}

// NewRegistryDirectory will create a new RegistryDirectory
func NewRegistryDirectory(url *common.URL, registry registry.Registry) (directory.Directory, error) {
	if url.SubURL == nil {
		return nil, perrors.Errorf("url is invalid, suburl can not be nil")
	}
	logger.Debugf("[Registry][Directory] new RegistryDirectory for service=%s", url.Key())

	commonConfig.EnsureApplicationAttribute(url, url.SubURL)
	ensureRegistriesAttribute(url)

	dir := &RegistryDirectory{
		Directory:           base.NewDirectory(url),
		cacheInvokers:       []protocolbase.Invoker{},
		cacheInvokersMap:    &sync.Map{},
		serviceType:         url.SubURL.Service(),
		registry:            registry,
		closingTombstones:   &sync.Map{},
		closingTombstoneTTL: defaultClosingTombstoneTTL,
	}

	dir.consumerURL = dir.getConsumerUrl(url.SubURL)

	if routerChain, err := chain.NewRouterChain(url); err == nil {
		dir.SetRouterChain(routerChain)
	} else {
		logger.Warnf("[Registry][Directory] fail to create router chain, url=%s err=%v", url.SubURL, err)
	}

	dir.consumerConfigurationListener = newConsumerConfigurationListener(dir, url)
	dir.consumerConfigurationListener.addNotifyListener(dir)
	dir.referenceConfigurationListener = newReferenceConfigurationListener(dir, url)
	graceful_shutdown.DefaultClosingDirectoryRegistry().Register(dir.closingServiceKey(), dir)

	if err := dir.registry.LoadSubscribeInstances(url.SubURL, dir); err != nil {
		return nil, err
	}
	metrics.Publish(metricsRegistry.NewDirectoryEvent(metricsRegistry.NumAllInc))
	return dir, nil
}

// subscribe from registry
func (dir *RegistryDirectory) Subscribe(url *common.URL) error {
	logger.Infof("[Registry][Directory] start subscribing for service=%s with a new go routine", url.Key())
	dir.setSubscribedURL(url)

	// Get the timeout time from the registration center configuration (default time 5s)
	registerUrl := dir.registry.GetURL()

	var timeoutStr string

	if registerUrl != nil {
		if val := registerUrl.GetParam(constant.RegistryTimeoutKey, ""); val != "" {
			timeoutStr = val
		}
	}

	timeout, err := time.ParseDuration(timeoutStr)
	if err != nil {
		logger.Warnf("[Registry][Directory] invalid timeout value=%s, using default=%s", timeoutStr, constant.DefaultRegTimeout)
		timeout, _ = time.ParseDuration(constant.DefaultRegTimeout)
	}

	serviceKey := url.Key()
	go func() {
		if err := dir.registry.Subscribe(url, dir); err != nil {
			logger.Errorf("[Registry][Directory] registry.Subscribe(url=%v dir=%v) = err=%v", url, dir, err)
		}
	}()

	// Registration is bounded by registry timeout (default 5s), but subscription
	// stays decoupled from registration so discovery can continue even on register error.
	if err := dir.registerConsumerWithTimeout(url, timeout, serviceKey); err != nil {
		return err
	}

	logger.Infof("[Registry][Directory] register completed successfully for service=%s", serviceKey)
	return nil
}

func (dir *RegistryDirectory) registerConsumerWithTimeout(url *common.URL, timeout time.Duration, serviceKey string) error {
	registerErrCh := make(chan error, 1)
	urlToReg := getConsumerUrlToRegistry(url.Clone())
	go func() {
		registerErrCh <- dir.registry.Register(urlToReg)
	}()

	timer := time.NewTimer(timeout)
	defer timer.Stop()

	select {
	case err := <-registerErrCh:
		if err != nil {
			registryURL := dir.registry.GetURL()
			registryURLString := ""
			if registryURL != nil {
				registryURLString = registryURL.String()
			}
			logger.Errorf("[Registry][Directory] consumer service %v register registry %v error, err=%v",
				url.String(), registryURLString, err)
			return err
		}
		return nil
	case <-timer.C:
		logger.Errorf("[Registry][Directory] register timed out for service=%s", serviceKey)
		go func() {
			err := <-registerErrCh
			if err != nil {
				return
			}
			if unRegErr := dir.registry.UnRegister(urlToReg.Clone()); unRegErr != nil {
				logger.Warnf("[Registry][Directory] register timed out for service=%s, but late unregister failed, err=%v", serviceKey, unRegErr)
			}
		}()
		return fmt.Errorf("register timed out for service: %s", serviceKey)
	}
}

// Notify monitor changes from registry,and update the cacheServices
func (dir *RegistryDirectory) Notify(event *registry.ServiceEvent) {
	if event == nil {
		return
	}
	start := time.Now()
	dir.refreshInvokers(event)
	metrics.Publish(metricsRegistry.NewNotifyEvent(start))
}

// NotifyAll notify the events that are complete Service Event List.
// After notify the address, the callback func will be invoked.
func (dir *RegistryDirectory) NotifyAll(events []*registry.ServiceEvent, callback func()) {
	dir.refreshAllInvokers(events, callback)
}

// refreshInvokers refreshes service's events.
func (dir *RegistryDirectory) refreshInvokers(event *registry.ServiceEvent) {
	if event != nil {
		logger.Debugf("[Registry][Directory] refresh invokers with %+v", event)
	} else {
		logger.Debug("[Registry][Directory] refresh invokers with nil")
	}

	var oldInvoker []protocolbase.Invoker
	if event != nil {
		oldInvoker, _ = dir.cacheInvokerByEvent(event)
	}
	dir.setNewInvokers()
	for _, v := range oldInvoker {
		if v != nil {
			v.Destroy()
		}
	}
}

// refreshAllInvokers the argument is the complete list of the service events,  we can safely assume any cached invoker
// not in the incoming list can be removed.  The Action of serviceEvent should be EventTypeUpdate or EventTypeAdd.
func (dir *RegistryDirectory) refreshAllInvokers(events []*registry.ServiceEvent, callback func()) {
	var (
		oldInvokers       []protocolbase.Invoker
		incomingProviders []*registry.ServiceEvent
		providerEvents    []*registry.ServiceEvent
		configuratorURLs  []*common.URL
	)
	for _, event := range events {
		if event.Action != remoting.EventTypeUpdate && event.Action != remoting.EventTypeAdd {
			panic("Your implements of register center is wrong, " +
				"please check the Action of ServiceEvent should be EventTypeUpdate")
		}
		if isConfiguratorURL(event.Service) {
			configuratorURLs = append(configuratorURLs, event.Service)
			continue
		}
		incomingProviders = append(incomingProviders, event)
	}
	if configuratorURLs != nil {
		dir.replaceConfigurators(configuratorURLs)
	}
	defer callback()
	if len(incomingProviders) > 0 {
		providerEvents = incomingProviders
		dir.cacheProviderEvents(providerEvents)
	} else if len(events) > 0 {
		providerEvents = dir.cachedProviderEventsSnapshot()
	} else {
		dir.cacheProviderEvents(nil)
	}
	if len(providerEvents) == 0 && len(events) > 0 {
		return
	}

	dir.overrideUrl(dir.GetDirectoryUrl())
	referenceUrl := dir.GetDirectoryUrl().SubURL

	// loop the events to check the Action should be EventTypeUpdate.
	for _, event := range providerEvents {
		// Originally it will Merge URL many times, now we just execute once.
		// MergeURL is executed once and put the result into Event. After this, the key will get from Event.Key().
		newUrl := dir.convertUrl(event)
		newUrl = newUrl.MergeURL(referenceUrl)
		dir.overrideUrl(newUrl)
		event.Update(newUrl)
	}
	func() {
		// this lock is work at batch update of InvokeCache
		dir.registerLock.Lock()
		defer dir.registerLock.Unlock()
		// get need clear invokers from original invoker list
		dir.cacheInvokersMap.Range(func(k, v any) bool {
			key, ok := k.(string)
			if !ok {
				logger.Errorf("[Registry][Directory] cached invoker has unexpected key type %T", k)
				dir.cacheInvokersMap.Delete(k)
				return true
			}
			if !dir.eventMatched(key, providerEvents) {
				// delete unused invoker from cache
				if invoker := dir.uncacheInvokerWithKey(key); invoker != nil {
					oldInvokers = append(oldInvokers, invoker)
				}
			}
			return true
		})
		// loop the serviceEvents
		for _, event := range providerEvents {
			logger.Debugf("[Registry][Directory] registry changed, result{%s}", event)
			if event != nil && event.Service != nil {
				logger.Infof("[Registry][Directory] selector add service url{%s}", event.Service.String())
			}
			if event != nil && event.Service != nil && constant.RouterProtocol == event.Service.Protocol {
				dir.configRouters()
			}
			if oldInvoker, _ := dir.doCacheInvoker(event.Service, event); oldInvoker != nil {
				oldInvokers = append(oldInvokers, oldInvoker)
			}
		}
	}()
	dir.setNewInvokers()
	// destroy unused invokers
	for _, invoker := range oldInvokers {
		go invoker.Destroy()
	}
}

// eventMatched checks if a cached invoker appears in the incoming invoker list, if no, then it is safe to remove.
func (dir *RegistryDirectory) eventMatched(key string, events []*registry.ServiceEvent) bool {
	for _, event := range events {
		if dir.invokerCacheKey(event) == key {
			return true
		}
	}
	return false
}

// invokerCacheKey generates the key in the cache for a given ServiceEvent.
func (dir *RegistryDirectory) invokerCacheKey(event *registry.ServiceEvent) string {
	// If the url is merged, then return Event.Key() directly.
	if event.Updated() {
		return event.Key()
	}
	referenceUrl := dir.GetDirectoryUrl().SubURL
	newUrl := event.Service.MergeURL(referenceUrl)
	event.Update(newUrl)
	return event.Key()
}

// setNewInvokers groups the invokers from the cache first, then set the result to both directory and router chain.
func (dir *RegistryDirectory) setNewInvokers() {
	newInvokers := dir.toGroupInvokers()
	dir.invokersLock.Lock()
	defer dir.invokersLock.Unlock()
	dir.cacheInvokers = newInvokers
	dir.RouterChain().SetInvokers(newInvokers)
}

// cacheInvokerByEvent caches invokers from the service event
func (dir *RegistryDirectory) cacheInvokerByEvent(event *registry.ServiceEvent) ([]protocolbase.Invoker, error) {
	// judge is override or others
	if event != nil {

		switch event.Action {
		case remoting.EventTypeAdd, remoting.EventTypeUpdate:
			u := dir.convertUrl(event)
			if u == nil && dir.cacheOriginUrl == nil {
				return nil, nil
			}
			logger.Infof("[Registry][Directory] selector add service url{%s}", event.Service)
			if u != nil && constant.RouterProtocol == u.Protocol {
				dir.configRouters()
			}
			return []protocolbase.Invoker{dir.cacheInvoker(u, event)}, nil
		case remoting.EventTypeDel:
			logger.Infof("[Registry][Directory] selector delete service url{%s}", event.Service)
			return dir.uncacheInvoker(event), nil
		default:
			return nil, fmt.Errorf("illegal event type: %v", event.Action)
		}
	}
	return nil, nil
}

// configRouters configures dynamic routers into the router chain, but, the current impl is incorrect, see FIXME above.
func (dir *RegistryDirectory) configRouters() {
}

// convertUrl processes override:// and router://
func (dir *RegistryDirectory) convertUrl(res *registry.ServiceEvent) *common.URL {
	ret := res.Service
	if ret.Protocol == constant.OverrideProtocol || // 1.for override url in 2.6.x
		ret.GetParam(constant.CategoryKey, constant.DefaultCategory) == constant.ConfiguratorsCategory {
		dir.appendConfigurator(extension.GetDefaultConfigurator(ret))
		ret = nil
	} else if ret.Protocol == constant.RouterProtocol || // 2.for router
		ret.GetParam(constant.CategoryKey, constant.DefaultCategory) == constant.RouterCategory {
		ret = nil
	}
	return ret
}

func isConfiguratorURL(url *common.URL) bool {
	return url != nil && (url.Protocol == constant.OverrideProtocol ||
		url.GetParam(constant.CategoryKey, constant.DefaultCategory) == constant.ConfiguratorsCategory)
}

func (dir *RegistryDirectory) replaceConfigurators(urls []*common.URL) {
	dir.configuratorsLock.Lock()
	defer dir.configuratorsLock.Unlock()
	dir.configurators = registry.ToConfigurators(urls, extension.GetDefaultConfiguratorFunc())
}

func (dir *RegistryDirectory) cacheProviderEvents(events []*registry.ServiceEvent) {
	dir.registerLock.Lock()
	defer dir.registerLock.Unlock()
	dir.cachedProviderEvents = cloneServiceEvents(events)
}

func (dir *RegistryDirectory) cachedProviderEventsSnapshot() []*registry.ServiceEvent {
	dir.registerLock.Lock()
	defer dir.registerLock.Unlock()
	return cloneServiceEvents(dir.cachedProviderEvents)
}

func cloneServiceEvents(events []*registry.ServiceEvent) []*registry.ServiceEvent {
	if len(events) == 0 {
		return nil
	}

	cloned := make([]*registry.ServiceEvent, 0, len(events))
	for _, event := range events {
		if event == nil {
			continue
		}
		clonedEvent := &registry.ServiceEvent{
			Action:  event.Action,
			KeyFunc: event.KeyFunc,
		}
		if event.Service != nil {
			clonedEvent.Service = event.Service.Clone()
		}
		cloned = append(cloned, clonedEvent)
	}
	return cloned
}

func (dir *RegistryDirectory) toGroupInvokers() []protocolbase.Invoker {
	groupInvokersMap := dir.groupInvokersFromCache()
	if len(groupInvokersMap) == 1 {
		for _, invokers := range groupInvokersMap {
			return invokers
		}
	}
	return dir.joinGroupInvokers(groupInvokersMap)
}

func (dir *RegistryDirectory) groupInvokersFromCache() map[string][]protocolbase.Invoker {
	groupInvokersMap := make(map[string][]protocolbase.Invoker)

	dir.cacheInvokersMap.Range(func(key, value any) bool {
		invoker, ok := cachedInvoker(key, value)
		if !ok {
			dir.cacheInvokersMap.Delete(key)
			return true
		}
		group := invoker.GetURL().GetParam(constant.GroupKey, "")
		groupInvokersMap[group] = append(groupInvokersMap[group], invoker)
		return true
	})
	return groupInvokersMap
}

func (dir *RegistryDirectory) joinGroupInvokers(groupInvokersMap map[string][]protocolbase.Invoker) []protocolbase.Invoker {
	groupInvokersList := make([]protocolbase.Invoker, 0, len(groupInvokersMap))
	for _, invokers := range groupInvokersMap {
		staticDir := static.NewDirectory(invokers)
		clusterKey := dir.GetURL().SubURL.GetParam(constant.ClusterKey, constant.DefaultCluster)
		cluster, err := extension.GetCluster(clusterKey)
		if err != nil {
			logger.Errorf("[Registry][Directory] directory get cluster %s error, err=%w, will skip this group",
				clusterKey, err)
			continue
		}
		if cluster == nil {
			logger.Errorf("[Registry][Directory] directory cluster is nil for key %s, will skip this group", clusterKey)
			continue
		}
		if err = staticDir.BuildRouterChain(invokers, dir.GetURL()); err != nil {
			logger.Errorf("[Registry][Directory] buildRouterChain error, err=%v", err)
			continue
		}
		groupInvokersList = append(groupInvokersList, cluster.Join(staticDir))
	}
	return groupInvokersList
}

func (dir *RegistryDirectory) uncacheInvokerWithClusterID(clusterID string) []protocolbase.Invoker {
	logger.Debugf("[Registry][Directory] all service will be deleted in cache invokers with clusterID=%s", clusterID)
	invokerKeys := make([]string, 0)
	dir.cacheInvokersMap.Range(func(key, cacheInvoker any) bool {
		invoker, ok := cachedInvoker(key, cacheInvoker)
		if !ok {
			dir.cacheInvokersMap.Delete(key)
			return true
		}
		keyString, ok := key.(string)
		if !ok {
			logger.Errorf("[Registry][Directory] cached invoker has unexpected key type %T", key)
			dir.cacheInvokersMap.Delete(key)
			return true
		}
		if invoker.GetURL().GetParam(constant.MeshClusterIDKey, "") == clusterID {
			invokerKeys = append(invokerKeys, keyString)
		}
		return true
	})
	uncachedInvokers := make([]protocolbase.Invoker, 0)
	for _, v := range invokerKeys {
		uncachedInvokers = append(uncachedInvokers, dir.uncacheInvokerWithKey(v))
	}
	return uncachedInvokers
}

// uncacheInvoker will return abandoned Invoker, if no Invoker to be abandoned, return nil
func (dir *RegistryDirectory) uncacheInvoker(event *registry.ServiceEvent) []protocolbase.Invoker {
	defer metrics.Publish(metricsRegistry.NewDirectoryEvent(metricsRegistry.NumDisableTotal))
	dir.clearClosingTombstone(event.Key())
	if clusterID := event.Service.GetParam(constant.MeshClusterIDKey, ""); event.Service.Location == constant.MeshAnyAddrMatcher && clusterID != "" {
		dir.uncacheInvokerWithClusterID(clusterID)
	}
	return []protocolbase.Invoker{dir.uncacheInvokerWithKey(event.Key())}
}

func (dir *RegistryDirectory) uncacheInvokerWithKey(key string) protocolbase.Invoker {
	logger.Debugf("[Registry][Directory] service will be deleted in cache invokers, key=%s", key)
	protocolbase.RemoveUrlKeyUnhealthyStatus(key)
	if cacheInvoker, ok := dir.cacheInvokersMap.Load(key); ok {
		dir.cacheInvokersMap.Delete(key)
		invoker, valid := cachedInvoker(key, cacheInvoker)
		if !valid {
			return nil
		}
		return invoker
	}
	return nil
}

// RemoveClosingInstance removes a single service instance from the directory by instanceKey.
// It is intended to be called by graceful shutdown logic before registry updates converge.
func (dir *RegistryDirectory) RemoveClosingInstance(instanceKey string) bool {
	if instanceKey == "" {
		return false
	}

	var removed protocolbase.Invoker
	func() {
		dir.registerLock.Lock()
		defer dir.registerLock.Unlock()

		if cacheInvoker, ok := dir.cacheInvokersMap.Load(instanceKey); ok {
			removed, _ = cachedInvoker(instanceKey, cacheInvoker)
		}
		dir.markClosingTombstone(instanceKey, removed, "closing-event")
		removed = dir.uncacheInvokerWithKey(instanceKey)
		if removed != nil {
			dir.setNewInvokers()
		}
	}()

	if removed != nil {
		removed.Destroy()
		return true
	}
	return false
}

func (dir *RegistryDirectory) markClosingTombstone(instanceKey string, invoker protocolbase.Invoker, source string) {
	if instanceKey == "" {
		return
	}

	tombstone := closingTombstone{
		InstanceKey: instanceKey,
		Source:      source,
		ExpireAt:    time.Now().Add(dir.closingTombstoneTTL),
	}
	if invoker != nil && invoker.GetURL() != nil {
		tombstone.ServiceKey = invoker.GetURL().ServiceKey()
		tombstone.Address = invoker.GetURL().Location
		tombstone.Timestamp = invoker.GetURL().GetParam(constant.TimestampKey, "")
	}
	dir.closingTombstones.Store(instanceKey, tombstone)
}

func (dir *RegistryDirectory) activeClosingTombstone(instanceKey string) (closingTombstone, bool) {
	if instanceKey == "" {
		return closingTombstone{}, false
	}
	tombstoneValue, ok := dir.closingTombstones.Load(instanceKey)
	if !ok {
		return closingTombstone{}, false
	}
	tombstone := tombstoneValue.(closingTombstone)
	if time.Now().After(tombstone.ExpireAt) {
		dir.closingTombstones.Delete(instanceKey)
		return closingTombstone{}, false
	}
	return tombstone, true
}

func (dir *RegistryDirectory) clearClosingTombstone(instanceKey string) {
	if instanceKey == "" {
		return
	}
	dir.closingTombstones.Delete(instanceKey)
}

func (dir *RegistryDirectory) cleanupExpiredClosingTombstones() {
	now := time.Now()
	dir.closingTombstones.Range(func(key, value any) bool {
		tombstone := value.(closingTombstone)
		if now.After(tombstone.ExpireAt) {
			dir.closingTombstones.Delete(key)
		}
		return true
	})
}

// cacheInvoker will return abandoned Invoker,if no Invoker to be abandoned,return nil
func (dir *RegistryDirectory) cacheInvoker(url *common.URL, event *registry.ServiceEvent) protocolbase.Invoker {
	dir.overrideUrl(dir.GetDirectoryUrl())
	referenceUrl := dir.GetDirectoryUrl().SubURL

	if url == nil && dir.cacheOriginUrl != nil {
		url = dir.cacheOriginUrl
	} else {
		dir.cacheOriginUrl = url
	}
	if url == nil {
		logger.Error("[Registry][Directory] url is nil, pls check if service url is subscribe successfully")
		return nil
	}
	// check the url's protocol is equal to the protocol which is configured in reference config or referenceUrl is not care about protocol
	if url.Protocol == referenceUrl.Protocol || referenceUrl.Protocol == "" {
		newUrl := url.MergeURL(referenceUrl)
		dir.overrideUrl(newUrl)
		event.Update(newUrl)
		if v, ok := dir.doCacheInvoker(newUrl, event); ok {
			return v
		}
	}
	return nil
}

func (dir *RegistryDirectory) doCacheInvoker(newUrl *common.URL, event *registry.ServiceEvent) (protocolbase.Invoker, bool) {
	key := event.Key()
	dir.cleanupExpiredClosingTombstones()
	if tombstone, ok := dir.activeClosingTombstone(key); ok {
		// A tombstone guards against re-adding an instance from a stale
		// pre-shutdown registry snapshot. If the re-add carries a different
		// export timestamp, the instance has genuinely restarted with the same
		// address: vetoing it would keep the directory empty until the next
		// registry event, which may never come.
		newTimestamp := newUrl.GetParam(constant.TimestampKey, "")
		if tombstone.Timestamp == "" || newTimestamp == "" || newTimestamp == tombstone.Timestamp {
			logger.Infof("[Registry][Directory] skip rebuilding closing instance due to tombstone, instance key: %s", key)
			return nil, true
		}
		logger.Infof("[Registry][Directory] instance %s restarted with a new export timestamp, clearing closing tombstone", key)
		dir.clearClosingTombstone(key)
	}
	cacheInvoker, ok := dir.cacheInvokersMap.Load(key)
	var existingInvoker protocolbase.Invoker
	if ok {
		existingInvoker, ok = cachedInvoker(key, cacheInvoker)
		if !ok {
			dir.cacheInvokersMap.Delete(key)
		}
	}
	if !ok {
		logger.Debugf("[Registry][Directory] service will be added in cache invokers, url=%s", newUrl)
		newInvoker := extension.GetProtocol(protocolwrapper.FILTER).Refer(newUrl)
		if newInvoker != nil {
			dir.cacheInvokersMap.Store(key, newInvoker)
		} else {
			logger.Warnf("[Registry][Directory] service will be added in cache invokers fail, result is null, url=%s", newUrl.String())
		}
	} else {
		metrics.Publish(metricsRegistry.NewDirectoryEvent(metricsRegistry.NumValidTotal))
		// if cached invoker has the same URL with the new URL, then no need to re-refer, and no need to destroy
		// the old invoker.
		if common.GetCompareURLEqualFunc()(newUrl, existingInvoker.GetURL()) {
			return nil, true
		}

		logger.Debugf("[Registry][Directory] service will be updated in cache invokers, newUrl=%s oldUrl=%s", newUrl, existingInvoker.GetURL())
		newInvoker := extension.GetProtocol(protocolwrapper.FILTER).Refer(newUrl)
		if newInvoker != nil {
			dir.cacheInvokersMap.Store(key, newInvoker)
			return existingInvoker, true
		} else {
			logger.Warnf("[Registry][Directory] service will be updated in cache invokers fail, result is null, url=%s", newUrl.String())
		}
	}
	return nil, false
}

func cachedInvoker(key, value any) (protocolbase.Invoker, bool) {
	invoker, ok := value.(protocolbase.Invoker)
	if !ok || isNilInvoker(invoker) {
		logger.Errorf("[Registry][Directory] cached invoker has unexpected type %T for key %v", value, key)
		return nil, false
	}
	return invoker, true
}

func isNilInvoker(invoker protocolbase.Invoker) bool {
	if invoker == nil {
		return true
	}
	value := reflect.ValueOf(invoker)
	switch value.Kind() {
	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice, reflect.UnsafePointer:
		return value.IsNil()
	default:
		return false
	}
}

// List selected protocol invokers from the directory
func (dir *RegistryDirectory) List(invocation protocolbase.Invocation) []protocolbase.Invoker {
	routerChain := dir.RouterChain()

	if routerChain == nil {
		return dir.snapshotCacheInvokers()
	}
	return routerChain.Route(dir.consumerURL, invocation)
}

// IsAvailable  whether the directory is available
func (dir *RegistryDirectory) IsAvailable() bool {
	if dir.IsDestroyed() {
		return false
	}

	for _, ivk := range dir.snapshotCacheInvokers() {
		if ivk.IsAvailable() {
			return true
		}
	}
	metrics.Publish(metricsRegistry.NewDirectoryEvent(metricsRegistry.NumToReconnectTotal))
	return false
}

// Destroy method
func (dir *RegistryDirectory) Destroy() {
	// TODO:unregister & unsubscribe
	dir.DoDestroy(func() {
		graceful_shutdown.DefaultClosingDirectoryRegistry().Unregister(dir.closingServiceKey(), dir)
		registeredURL, subscribedURL := dir.snapshotRegistryURLs()

		if registeredURL != nil {
			err := dir.registry.UnRegister(registeredURL)
			if err != nil {
				logger.Warnf("[Registry][Directory] unregister consumer url failed, url=%s err=%v", registeredURL.String(), err)
			}
		}

		if subscribedURL != nil {
			err := dir.registry.UnSubscribe(subscribedURL, dir)
			if err != nil {
				logger.Warnf("[Registry][Directory] unsubscribe consumer url failed, url=%s err=%v", subscribedURL.String(), err)
			}
		}

		invokers := dir.swapCacheInvokers()
		for _, ivk := range invokers {
			ivk.Destroy()
		}
		dir.cacheInvokersMap.Range(func(key, value any) bool {
			dir.cacheInvokersMap.Delete(key)
			return true
		})
		dir.registerLock.Lock()
		dir.cachedProviderEvents = nil
		dir.registerLock.Unlock()
	})
	metrics.Publish(metricsRegistry.NewDirectoryEvent(metricsRegistry.NumAllDec))
}

func (dir *RegistryDirectory) closingServiceKey() string {
	if dir.GetURL() == nil {
		return ""
	}
	serviceKey := dir.GetURL().ServiceKey()
	if serviceKey == "" && dir.GetURL().SubURL != nil {
		serviceKey = dir.GetURL().SubURL.ServiceKey()
	}
	return serviceKey
}

func (dir *RegistryDirectory) overrideUrl(targetUrl *common.URL) {
	// Use a read-only snapshot to avoid sharing mutable configurator slice during overrides.
	doOverrideUrl(dir.snapshotConfigurators(), targetUrl)
	doOverrideUrl(dir.consumerConfigurationListener.Configurators(), targetUrl)
	doOverrideUrl(dir.referenceConfigurationListener.Configurators(), targetUrl)
}

func (dir *RegistryDirectory) snapshotCacheInvokers() []protocolbase.Invoker {
	dir.invokersLock.RLock()
	defer dir.invokersLock.RUnlock()
	invokers := make([]protocolbase.Invoker, len(dir.cacheInvokers))
	copy(invokers, dir.cacheInvokers)
	return invokers
}

func (dir *RegistryDirectory) swapCacheInvokers() []protocolbase.Invoker {
	dir.invokersLock.Lock()
	defer dir.invokersLock.Unlock()
	invokers := make([]protocolbase.Invoker, len(dir.cacheInvokers))
	copy(invokers, dir.cacheInvokers)
	dir.cacheInvokers = []protocolbase.Invoker{}
	return invokers
}

func (dir *RegistryDirectory) appendConfigurator(configurator config_center.Configurator) {
	dir.configuratorsLock.Lock()
	defer dir.configuratorsLock.Unlock()
	dir.configurators = append(dir.configurators, configurator)
}

func (dir *RegistryDirectory) snapshotConfigurators() []config_center.Configurator {
	dir.configuratorsLock.RLock()
	defer dir.configuratorsLock.RUnlock()
	configurators := make([]config_center.Configurator, len(dir.configurators))
	copy(configurators, dir.configurators)
	return configurators
}

func (dir *RegistryDirectory) setSubscribedURL(url *common.URL) {
	dir.subscribedUrlLock.Lock()
	defer dir.subscribedUrlLock.Unlock()
	dir.SubscribedUrl = url
}

func (dir *RegistryDirectory) snapshotRegistryURLs() (registeredURL *common.URL, subscribedURL *common.URL) {
	dir.subscribedUrlLock.RLock()
	defer dir.subscribedUrlLock.RUnlock()
	return dir.RegisteredUrl, dir.SubscribedUrl
}

func (dir *RegistryDirectory) getConsumerUrl(c *common.URL) *common.URL {
	processID := fmt.Sprintf("%d", os.Getpid())
	localIP := common.GetLocalIp()

	params := url.Values{}
	c.RangeParams(func(key, value string) bool {
		params.Add(key, value)
		return true
	})

	params.Add("pid", processID)
	params.Add("ip", localIP)
	params.Add("protocol", c.Protocol)

	return common.NewURLWithOptions(common.WithProtocol("consumer"), common.WithIp(localIP), common.WithPath(c.Path),
		common.WithParams(params))
}

func doOverrideUrl(configurators []config_center.Configurator, targetUrl *common.URL) {
	for _, v := range configurators {
		v.Configure(targetUrl)
	}
}

type referenceConfigurationListener struct {
	registry.BaseConfigurationListener
	directory *RegistryDirectory
	url       *common.URL
}

func newReferenceConfigurationListener(dir *RegistryDirectory, url *common.URL) *referenceConfigurationListener {
	listener := &referenceConfigurationListener{directory: dir, url: url}
	listener.InitWith(
		url.ColonSeparatedKey()+constant.ConfiguratorSuffix,
		listener,
		extension.GetDefaultConfiguratorFunc(),
	)
	return listener
}

// Process handle events and update Invokers
func (l *referenceConfigurationListener) Process(event *config_center.ConfigChangeEvent) {
	l.BaseConfigurationListener.Process(event)
	// FIXME: this doesn't trigger dir.overrideUrl()
	l.directory.refreshInvokers(nil)
}

type consumerConfigurationListener struct {
	registry.BaseConfigurationListener
	listeners []registry.NotifyListener
	directory *RegistryDirectory
}

func newConsumerConfigurationListener(dir *RegistryDirectory, url *common.URL) *consumerConfigurationListener {
	listener := &consumerConfigurationListener{directory: dir}
	application := commonConfig.EnsureApplicationAttribute(url, url.SubURL)
	listener.InitWith(
		application.Name+constant.ConfiguratorSuffix,
		listener,
		extension.GetDefaultConfiguratorFunc(),
	)

	return listener
}

func (l *consumerConfigurationListener) addNotifyListener(listener registry.NotifyListener) {
	l.listeners = append(l.listeners, listener)
}

// Process handles events from Configuration Center and update Invokers
func (l *consumerConfigurationListener) Process(event *config_center.ConfigChangeEvent) {
	l.BaseConfigurationListener.Process(event)
	// FIXME: this doesn't trigger dir.overrideUrl()
	l.directory.refreshInvokers(nil)
}

// ServiceDiscoveryRegistryDirectory implementation of Directory:
// Invoker list returned from this Directory's list method have been filtered by Routers
type ServiceDiscoveryRegistryDirectory struct {
	*base.Directory
	*RegistryDirectory
}

// NewServiceDiscoveryRegistryDirectory will create a new ServiceDiscoveryRegistryDirectory
func NewServiceDiscoveryRegistryDirectory(url *common.URL, registry registry.Registry) (directory.Directory, error) {
	dic, err := NewRegistryDirectory(url, registry)
	registryDirectory, _ := dic.(*RegistryDirectory)
	return &ServiceDiscoveryRegistryDirectory{
		Directory:         registryDirectory.Directory,
		RegistryDirectory: registryDirectory,
	}, err
}

// Subscribe do subscribe from registry
func (dir *ServiceDiscoveryRegistryDirectory) Subscribe(url *common.URL) error {
	if err := dir.registry.Subscribe(url, dir); err != nil {
		logger.Errorf("[Registry][Directory] registry.Subscribe(url=%v dir=%v) = err=%v", url, dir, err)
		return err
	}

	urlToReg := getConsumerUrlToRegistry(url)
	err := dir.registry.Register(urlToReg)
	if err != nil {
		logger.Errorf("[Registry][Directory] consumer service %v register registry %v error, err=%v",
			url.String(), dir.registry.GetURL().String(), err)
		return err
	}
	return nil
}

// List selected protocol invokers from the directory
func (dir *ServiceDiscoveryRegistryDirectory) List(invocation protocolbase.Invocation) []protocolbase.Invoker {
	return dir.RegistryDirectory.List(invocation)
}

func getConsumerUrlToRegistry(url *common.URL) *common.URL {
	// if developer define registry port and ip, use it first.
	if ipToRegistry := os.Getenv(constant.DubboIpToRegistryKey); len(ipToRegistry) > 0 {
		url.Ip = ipToRegistry
	} else {
		url.Ip = common.GetLocalIp()
	}
	if portToRegistry := os.Getenv(constant.DubboPortToRegistryKey); len(portToRegistry) > 0 {
		url.Port = portToRegistry
	}
	return url
}