feat(registry): consul support re-registry when service does not exist (#2606)

contrib/registry/consul/client.go

@@ -3,6 +3,7 @@ package consul
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"net"
 	"net/url"
 	"strconv"
@@ -41,25 +42,6 @@ type Client struct {
 	serviceChecks api.AgentServiceChecks
 }
 
-// Deprecated use newClient instead.
-func NewClient(cli *api.Client) *Client {
-	return newClient(cli, SingleDatacenter)
-}
-
-func newClient(cli *api.Client, dc Datacenter) *Client {
-	c := &Client{
-		dc:                             dc,
-		cli:                            cli,
-		resolver:                       defaultResolver,
-		healthcheckInterval:            10,
-		heartbeat:                      true,
-		deregisterCriticalServiceAfter: 600,
-	}
-
-	c.ctx, c.cancel = context.WithCancel(context.Background())
-	return c
-}
-
 func defaultResolver(_ context.Context, entries []*api.ServiceEntry) []*registry.ServiceInstance {
 	services := make([]*registry.ServiceInstance, 0, len(entries))
 	for _, entry := range entries {
@@ -225,7 +207,15 @@ func (c *Client) Register(_ context.Context, svc *registry.ServiceInstance, enab
 				case <-ticker.C:
 					err = c.cli.Agent().UpdateTTL("service:"+svc.ID, "pass", "pass")
 					if err != nil {
-						log.Errorf("[Consul]update ttl heartbeat to consul failed!err:=%v", err)
+						log.Errorf("[Consul] update ttl heartbeat to consul failed! err=%v", err)
+						// when the previous report fails, try to re register the service
+						time.AfterFunc(time.Duration(rand.Intn(5))*time.Second, func() {
+							if err := c.cli.Agent().ServiceRegister(asr); err != nil {
+								log.Errorf("[Consul] re registry service failed!, err=%v", err)
+							} else {
+								log.Warn("[Consul] re registry of service occurred success")
+							}
+						})
 					}
 				case <-c.ctx.Done():
 					return

contrib/registry/consul/registry.go

@@ -37,7 +37,7 @@ func WithTimeout(timeout time.Duration) Option {
 // WithDatacenter with registry datacenter option
 func WithDatacenter(dc Datacenter) Option {
 	return func(o *Registry) {
-		o.dc = dc
+		o.cli.dc = dc
 	}
 }
 
@@ -98,21 +98,27 @@ type Registry struct {
 	registry          map[string]*serviceSet
 	lock              sync.RWMutex
 	timeout           time.Duration
-	dc                Datacenter
 }
 
 // New creates consul registry
 func New(apiClient *api.Client, opts ...Option) *Registry {
 	r := &Registry{
-		dc:                SingleDatacenter,
 		registry:          make(map[string]*serviceSet),
 		enableHealthCheck: true,
 		timeout:           10 * time.Second,
+		cli: &Client{
+			dc:                             SingleDatacenter,
+			cli:                            apiClient,
+			resolver:                       defaultResolver,
+			healthcheckInterval:            10,
+			heartbeat:                      true,
+			deregisterCriticalServiceAfter: 600,
+		},
 	}
 	for _, o := range opts {
 		o(r)
 	}
-	r.cli = newClient(apiClient, r.dc)
+	r.cli.ctx, r.cli.cancel = context.WithCancel(context.Background())
 	return r
 }