Consul name resolution in-memory cache

nameresolution/consul/README.md

@@ -54,7 +54,8 @@ As of writing the configuration spec is fixed to v1.3.0 of the consul api
 | Tags | `[]string` | Configures any tags to include if/when registering services |
 | Meta | `map[string]string` | Configures any additional metadata to include if/when registering services |
 | DaprPortMetaKey | `string` | The key used for getting the Dapr sidecar port from consul service metadata during service resolution, it will also be used to set the Dapr sidecar port in metadata during registration. If blank it will default to `DAPR_PORT` |
-| SelfRegister | `bool` | Controls if Dapr will register the service to consul. The name resolution interface does not cater for an "on shutdown" pattern so please consider this if using Dapr to register services to consul as it will not deregister services. |
+| SelfRegister | `bool` | Controls if Dapr will register the service to consul on startup. If unset it will default to `false` |
+| SelfDeregister | `bool` | Controls if Dapr will deregister the service from consul on shutdown. If unset it will default to `false` |
 | AdvancedRegistration | [*api.AgentServiceRegistration](https://pkg.go.dev/github.com/hashicorp/consul/api@v1.3.0#AgentServiceRegistration) | Gives full control of service registration through configuration. If configured the component will ignore any configuration of Checks, Tags, Meta and SelfRegister. |
 | UseCache | `bool` | Configures if Dapr will cache the resolved services in-memory. This is done using consul [blocking queries](https://www.consul.io/api-docs/features/blocking) which can be configured via the QueryOptions configuration. If unset it will default to `false` |
 ## Samples Configurations

nameresolution/consul/configuration.go

@@ -37,6 +37,7 @@ type intermediateConfig struct {
 	AdvancedRegistration *AgentServiceRegistration // advanced use-case
 	DaprPortMetaKey      string
 	SelfRegister         bool
+	SelfDeregister       bool
 	UseCache             bool
 }
 
@@ -49,6 +50,7 @@ type configSpec struct {
 	AdvancedRegistration *consul.AgentServiceRegistration // advanced use-case
 	DaprPortMetaKey      string
 	SelfRegister         bool
+	SelfDeregister       bool
 	UseCache             bool
 }
 
@@ -90,6 +92,7 @@ func mapConfig(config intermediateConfig) configSpec {
 		QueryOptions:         mapQueryOptions(config.QueryOptions),
 		AdvancedRegistration: mapAdvancedRegistration(config.AdvancedRegistration),
 		SelfRegister:         config.SelfRegister,
+		SelfDeregister:       config.SelfDeregister,
 		DaprPortMetaKey:      config.DaprPortMetaKey,
 		UseCache:             config.UseCache,
 	}

nameresolution/consul/consul.go

@@ -19,6 +19,7 @@ import (
 	"net"
 	"strconv"
 	"sync"
+	"sync/atomic"
 
 	consul "github.com/hashicorp/consul/api"
 
@@ -58,6 +59,7 @@ type clientInterface interface {
 type agentInterface interface {
 	Self() (map[string]map[string]interface{}, error)
 	ServiceRegister(service *consul.AgentServiceRegistration) error
+	ServiceDeregister(serviceID string) error
 }
 
 type healthInterface interface {
@@ -66,12 +68,12 @@ type healthInterface interface {
 }
 
 type resolver struct {
-	config         resolverConfig
-	logger         logger.Logger
-	client         clientInterface
-	registry       registryInterface
-	watcherStarted bool
-	watcherMutex   sync.Mutex
+	config             resolverConfig
+	logger             logger.Logger
+	client             clientInterface
+	registry           registryInterface
+	watcherStarted     atomic.Bool
+	watcherStopChannel chan struct{}
 }
 
 type registryInterface interface {
@@ -86,7 +88,7 @@ type registryInterface interface {
 }
 
 type registry struct {
-	entries        *sync.Map
+	entries        sync.Map
 	serviceChannel chan string
 }
 
@@ -130,9 +132,7 @@ func (r *resolver) getService(service string) (*consul.ServiceEntry, error) {
 	var services []*consul.ServiceEntry
 
 	if r.config.UseCache {
-		if !r.watcherStarted {
-			r.startWatcher()
-		}
+		r.startWatcher()
 
 		entry := r.registry.get(service)
 		if entry != nil {
@@ -214,24 +214,26 @@ func (r *registry) registrationChannel() chan string {
 }
 
 type resolverConfig struct {
-	Client          *consul.Config
-	QueryOptions    *consul.QueryOptions
-	Registration    *consul.AgentServiceRegistration
-	DaprPortMetaKey string
-	UseCache        bool
+	Client            *consul.Config
+	QueryOptions      *consul.QueryOptions
+	Registration      *consul.AgentServiceRegistration
+	DeregisterOnClose bool
+	DaprPortMetaKey   string
+	UseCache          bool
 }
 
 // NewResolver creates Consul name resolver.
 func NewResolver(logger logger.Logger) nr.Resolver {
-	return newResolver(logger, resolverConfig{}, &client{}, &registry{entries: &sync.Map{}, serviceChannel: make(chan string, 100)})
+	return newResolver(logger, resolverConfig{}, &client{}, &registry{serviceChannel: make(chan string, 100)}, make(chan struct{}))
 }
 
-func newResolver(logger logger.Logger, resolverConfig resolverConfig, client clientInterface, registry registryInterface) nr.Resolver {
+func newResolver(logger logger.Logger, resolverConfig resolverConfig, client clientInterface, registry registryInterface, watcherStopChannel chan struct{}) nr.Resolver {
 	return &resolver{
-		logger:   logger,
-		config:   resolverConfig,
-		client:   client,
-		registry: registry,
+		logger:             logger,
+		config:             resolverConfig,
+		client:             client,
+		registry:           registry,
+		watcherStopChannel: watcherStopChannel,
 	}
 }
 
@@ -295,6 +297,24 @@ func (r *resolver) ResolveID(req nr.ResolveRequest) (addr string, err error) {
 	return formatAddress(addr, port)
 }
 
+// Close will stop the watcher and deregister app from consul
+func (r *resolver) Close() error {
+	if r.watcherStarted.Load() {
+		r.watcherStopChannel <- struct{}{}
+	}
+
+	if r.config.Registration != nil && r.config.DeregisterOnClose {
+		err := r.client.Agent().ServiceDeregister(r.config.Registration.ID)
+		if err != nil {
+			return fmt.Errorf("failed to deregister consul service: %w", err)
+		}
+
+		r.logger.Info("deregistered service from consul")
+	}
+
+	return nil
+}
+
 func formatAddress(address string, port string) (addr string, err error) {
 	if net.ParseIP(address).To4() != nil {
 		return address + ":" + port, nil
@@ -317,6 +337,7 @@ func getConfig(metadata nr.Metadata) (resolverCfg resolverConfig, err error) {
 	}
 
 	resolverCfg.DaprPortMetaKey = cfg.DaprPortMetaKey
+	resolverCfg.DeregisterOnClose = cfg.SelfDeregister
 	resolverCfg.UseCache = cfg.UseCache
 
 	resolverCfg.Client = getClientConfig(cfg)