update

myregistry/consul/builder.go

@@ -1,200 +0,0 @@
-package consul
-
-import (
-	"context"
-	"fmt"
-	"github.com/jpillora/backoff"
-	"google.golang.org/grpc/grpclog"
-	"log"
-	"sort"
-	"strings"
-	"time"
-
-	"github.com/hashicorp/consul/api"
-	"github.com/pkg/errors"
-	"google.golang.org/grpc/resolver"
-)
-
-// schemeName for the urls
-// All target URLs like 'consul://.../...' will be resolved by this resolver
-const schemeName = "consul"
-
-// builder implements resolver.Builder and use for constructing all consul resolvers
-type builder struct{}
-
-func (b *builder) Build(url resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
-	tgt, err := parseURL(url.URL.String())
-	if err != nil {
-		return nil, errors.Wrap(err, "Wrong consul URL")
-	}
-	cli, err := api.NewClient(tgt.consulConfig())
-	if err != nil {
-		return nil, errors.Wrap(err, "Couldn't connect to the Consul API")
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	pipe := make(chan []string)
-	go watchConsulService(ctx, cli.Health(), tgt, pipe)
-	go populateEndpoints(ctx, cc, pipe, tgt)
-
-	return &resolvr{cancelFunc: cancel}, nil
-}
-
-// Scheme returns the scheme supported by this resolver.
-// Scheme is defined at https://github.com/grpc/grpc/blob/master/doc/naming.md.
-func (b *builder) Scheme() string {
-	return schemeName
-}
-
-// init function needs for  auto-register in resolvers registry
-func init() {
-	resolver.Register(&builder{})
-}
-
-// resolvr implements resolver.Resolver from the gRPC package.
-// It watches for endpoints changes and pushes them to the underlying gRPC connection.
-type resolvr struct {
-	cancelFunc context.CancelFunc
-}
-
-// ResolveNow will be skipped due unnecessary in this case
-func (r *resolvr) ResolveNow(resolver.ResolveNowOptions) {}
-
-// Close closes the resolver.
-func (r *resolvr) Close() {
-	r.cancelFunc()
-}
-
-//go:generate ./bin/moq -out mocks_test.go . servicer
-type servicer interface {
-	Service(string, string, bool, *api.QueryOptions) ([]*api.ServiceEntry, *api.QueryMeta, error)
-}
-
-func watchConsulService(ctx context.Context, s servicer, tgt target, out chan<- []string) {
-	res := make(chan []string)
-	quit := make(chan struct{})
-	bck := &backoff.Backoff{
-		Factor: 2,
-		Jitter: true,
-		Min:    10 * time.Millisecond,
-		Max:    tgt.MaxBackoff,
-	}
-	go func() {
-		var lastIndex uint64
-		for {
-			ss, meta, err := s.Service(
-				tgt.Service,
-				tgt.Tag,
-				tgt.Healthy,
-				&api.QueryOptions{
-					WaitIndex:         lastIndex,
-					Near:              tgt.Near,
-					WaitTime:          tgt.Wait,
-					Datacenter:        tgt.Dc,
-					AllowStale:        tgt.AllowStale,
-					RequireConsistent: tgt.RequireConsistent,
-				},
-			)
-			if err != nil {
-				// No need to continue if the context is done/cancelled.
-				// We check that here directly because the check for the closed quit channel
-				// at the end of the loop is not reached when calling continue here.
-				select {
-				case <-quit:
-					return
-				default:
-					grpclog.Errorf("[Consul resolver] Couldn't fetch endpoints. target={%s}; error={%v}", tgt.String(), err)
-					time.Sleep(bck.Duration())
-					continue
-				}
-			}
-			bck.Reset()
-			lastIndex = meta.LastIndex
-			grpclog.Infof("[Consul resolver] %d endpoints fetched in(+wait) %s for target={%s}",
-				len(ss),
-				meta.RequestTime,
-				tgt.String(),
-			)
-
-			ee := make([]string, 0, len(ss))
-			for _, s := range ss {
-				address := s.Service.Address
-				if s.Service.Address == "" {
-					address = s.Node.Address
-				}
-				ee = append(ee, fmt.Sprintf("%s:%d", address, s.Service.Port))
-			}
-
-			if tgt.Limit != 0 && len(ee) > tgt.Limit {
-				ee = ee[:tgt.Limit]
-			}
-			select {
-			case res <- ee:
-				continue
-			case <-quit:
-				return
-			}
-		}
-	}()
-
-	for {
-		// If in the below select both channels have values that can be read,
-		// Go picks one pseudo-randomly.
-		// But when the context is canceled we want to act upon it immediately.
-		if ctx.Err() != nil {
-			// Close quit so the goroutine returns and doesn't leak.
-			// Do NOT close res because that can lead to panics in the goroutine.
-			// res will be garbage collected at some point.
-			close(quit)
-			return
-		}
-		select {
-		case ee := <-res:
-			out <- ee
-		case <-ctx.Done():
-			close(quit)
-			return
-		}
-	}
-}
-
-func populateEndpoints(ctx context.Context, clientConn resolver.ClientConn, input <-chan []string, tgt target) {
-	for {
-		select {
-		case cc := <-input:
-			connsSet := make(map[string]struct{}, len(cc))
-			for _, c := range cc {
-				connsSet[c] = struct{}{}
-			}
-			conns := make([]resolver.Address, 0, len(connsSet))
-			for c := range connsSet {
-				conns = append(conns, resolver.Address{Addr: c})
-			}
-
-			sort.Sort(byAddressString(conns)) // Don't replace the same address list in the balancer
-
-			connInfo := make([]string, 0, len(conns))
-			for _, conn := range conns {
-				connInfo = append(connInfo, conn.Addr)
-			}
-
-			log.Printf("update conn: %s num:%d [%s]", tgt.Service, len(connInfo), strings.Join(connInfo, ", "))
-
-			err := clientConn.UpdateState(resolver.State{Addresses: conns})
-			if err != nil {
-				grpclog.Errorf("[Consul resolver %s] Couldn't update client connection. error={%v}", tgt.Service, err)
-				continue
-			}
-		case <-ctx.Done():
-			grpclog.Infof("[Consul resolver %s] Watch has been finished", tgt.Service)
-			return
-		}
-	}
-}
-
-// byAddressString sorts resolver.Address by Address Field  sorting in increasing order.
-type byAddressString []resolver.Address
-
-func (p byAddressString) Len() int           { return len(p) }
-func (p byAddressString) Less(i, j int) bool { return p[i].Addr < p[j].Addr }
-func (p byAddressString) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }

myregistry/consul/consul.go

@@ -4,19 +4,30 @@ import (
 	"fmt"
 	"git.makemake.in/kzkzzzz/mycommon/myconf"
 	"git.makemake.in/kzkzzzz/mycommon/mygrpc"
+	"git.makemake.in/kzkzzzz/mycommon/mylog"
 	"git.makemake.in/kzkzzzz/mycommon/myregistry"
 	"github.com/hashicorp/consul/api"
-	"github.com/hashicorp/consul/api/watch"
-	"github.com/hashicorp/go-hclog"
 	"github.com/rs/xid"
-	"log"
+	"go.uber.org/zap"
 	"net"
 	"net/url"
 	"os"
+	"sync"
 	"time"
 )
 
-var _ myregistry.IRegister = (*Consul)(nil)
+var (
+	_ myregistry.IRegister = (*Consul)(nil)
+
+	defaultLog = mylog.NewLogger(&mylog.Config{
+		Level:         mylog.DebugLevel,
+		NeedLogFile:   false,
+		ConsoleWriter: os.Stdout,
+		ZapOpt: []zap.Option{
+			zap.AddCaller(), zap.AddCallerSkip(1),
+		},
+	})
+)
 
 type Consul struct {
 	client      *api.Client
@@ -42,7 +53,7 @@ func (c *Consul) Register(service *myregistry.ServiceInfo) error {
 	serviceId := xid.New().String()
 	hostname, err := os.Hostname()
 	if err != nil {
-		log.Printf("get hostname err: %s", err)
+		defaultLog.Errorf("get hostname err: %s", err)
 	} else {
 		serviceId = fmt.Sprintf("%s-%s", hostname, serviceId)
 	}
@@ -57,12 +68,20 @@ func (c *Consul) Register(service *myregistry.ServiceInfo) error {
 		// 指定时间后自动注销不健康的服务节点
 		// 最小超时时间为1分钟，收获不健康服务的进程每30秒运行一次，因此触发注销的时间可能略长于配置的超时时间。
 		DeregisterCriticalServiceAfter: "6m",
-		Status:                         "passing",
+		Status:                         api.HealthPassing,
 	}
+
+	regTags := make([]string, len(c.serviceTags))
+	copy(regTags, c.serviceTags)
+
+	if v := service.Extend["tag"]; v != "" {
+		regTags = append(regTags, v)
+	}
+
 	svc := &api.AgentServiceRegistration{
 		ID:      serviceId,           // 服务唯一ID
 		Name:    service.ServiceName, // 服务名称
-		Tags:    c.serviceTags,       // 为服务打标签
+		Tags:    regTags,             // 为服务打标签
 		Address: service.Ip,
 		Port:    service.Port,
 		Check:   check,
@@ -70,14 +89,22 @@ func (c *Consul) Register(service *myregistry.ServiceInfo) error {
 
 	c.services = append(c.services, svc)
 
-	return c.client.Agent().ServiceRegister(svc)
+	err = c.client.Agent().ServiceRegister(svc)
+	if err != nil {
+		defaultLog.Errorf("retry register service err: %s: %s", svc.Name, err)
+		return err
+	}
+
+	defaultLog.Infof("retry register service ok: %s", svc.Name)
+
+	return nil
 }
 
 func (c *Consul) Deregister(service *myregistry.ServiceInfo) error {
 	for _, svcId := range c.serviceIds[service.ServiceName] {
 		err := c.client.Agent().ServiceDeregister(svcId)
 		if err != nil {
-			log.Printf("Failed to deregister service %s: %s\n", service, err)
+			defaultLog.Errorf("Failed to deregister service %s: %s\n", service, err)
 		}
 	}
 	return nil
@@ -92,32 +119,12 @@ func MustNew(conf *myconf.Config, opts ...Opt) *Consul {
 }
 
 func New(conf *myconf.Config, opts ...Opt) (*Consul, error) {
-
-	cfg := api.DefaultConfig()
-	cfg.Address = conf.GetString("addr")
-
-	if cfg.Address == "" {
-		return nil, fmt.Errorf("consul address is empty")
-	}
-
-	cfg.Transport.DialContext = (&net.Dialer{
-		Timeout:   3 * time.Second,
-		KeepAlive: 20 * time.Second,
-		DualStack: true,
-	}).DialContext
-	cfg.Token = conf.GetString("token")
-
-	username := conf.GetString("username")
-	password := conf.GetString("password")
-
-	if username != "" && password != "" {
-		cfg.HttpAuth = &api.HttpBasicAuth{
-			Username: username,
-			Password: password,
-		}
-	}
-
-	client, err := api.NewClient(cfg)
+	client, err := NewClient(&ClientConfig{
+		Address:  conf.GetString("addr"),
+		Token:    conf.GetString("token"),
+		Username: conf.GetString("username"),
+		Password: conf.GetString("password"),
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -138,39 +145,108 @@ func New(conf *myconf.Config, opts ...Opt) (*Consul, error) {
 	return cl, nil
 }
 
-func (c *Consul) healthCheck() {
-	wlog := newWatchLogger()
+type ClientConfig struct {
+	Address        string
+	Token          string
+	Username       string
+	Password       string
+	AutoGrpcPrefix bool
+}
 
-	wp, err := watch.Parse(map[string]any{
-		"type": "services",
-	})
-	if err != nil {
-		panic(fmt.Sprintf("parse watch err: %s", err))
+var (
+	clientLock = &sync.Mutex{}
+	clientMap  = make(map[string]*api.Client)
+)
+
+func NewClient(clientCfg *ClientConfig) (*api.Client, error) {
+	clientLock.Lock()
+	defer clientLock.Unlock()
+
+	if clientCfg.Address == "" {
+		return nil, fmt.Errorf("consul address is empty")
 	}
 
-	wp.Handler = func(u uint64, raw any) {
-		if wlog.isWatchErr == true {
+	if client, ok := clientMap[clientCfg.Address]; ok {
+		return client, nil
+	}
 
-			for _, svc := range c.services {
-				//c.client.Agent().ServiceDeregister(svc.ID)
-				err := c.client.Agent().ServiceRegister(svc)
-				if err != nil {
-					log.Printf("retry register service err: %s: %s", svc.Name, err)
-				} else {
-					log.Printf("retry register service ok: %s", svc.Name)
+	cfg := api.DefaultConfig()
+	cfg.Address = clientCfg.Address
+
+	if cfg.Address == "" {
+		return nil, fmt.Errorf("consul address is empty")
+	}
+
+	cfg.Transport.DialContext = (&net.Dialer{
+		Timeout:   3 * time.Second,
+		KeepAlive: 20 * time.Second,
+	}).DialContext
+	cfg.Token = clientCfg.Token
+
+	username := clientCfg.Username
+	password := clientCfg.Password
+
+	if username != "" && password != "" {
+		cfg.HttpAuth = &api.HttpBasicAuth{
+			Username: username,
+			Password: password,
+		}
+	}
+
+	client, err := api.NewClient(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	clientMap[clientCfg.Address] = client
+	return client, nil
+}
+
+func (c *Consul) healthCheck() {
+	var (
+		lastIndex uint64
+		isFirst   = true
+		err       error
+		meta      *api.QueryMeta
+	)
+
+	for {
+		if isFirst == false {
+			time.Sleep(time.Second * 2)
+
+			// 错误的情况重新注册一次
+			if err != nil {
+				var isRegisterErr bool
+				for _, svc := range c.services {
+					err := c.client.Agent().ServiceRegister(svc)
+					if err != nil {
+						defaultLog.Errorf("retry register service err: %s: %s", svc.Name, err)
+						isRegisterErr = true
+						break
+					} else {
+						defaultLog.Infof("retry register service ok: %s", svc.Name)
+					}
+				}
+
+				if isRegisterErr {
+					continue
 				}
 			}
 
-			wlog.isWatchErr = false
 		}
-		//fmt.Println("watch", u, raw)
-	}
 
-	err = wp.RunWithClientAndHclog(c.client, wlog)
-	if err != nil {
-		log.Printf("watch err: %s", err)
-	}
+		isFirst = false
 
+		_, meta, err = c.client.Catalog().Nodes(&api.QueryOptions{
+			WaitIndex: lastIndex,
+			//WaitTime:  time.Second * 5,
+		})
+		if err != nil {
+			defaultLog.Errorf("health check err: %s", err)
+			continue
+		}
+		lastIndex = meta.LastIndex
+	}
 }
 
 func (c *Consul) Client() *api.Client {
@@ -178,13 +254,27 @@ func (c *Consul) Client() *api.Client {
 }
 
 func GrpcUrl(serviceName string, conf *myconf.Config) string {
-	return GrpcUrlWithTag("", mygrpc.ServicePrefix+serviceName, conf)
+	return GrpcUrlWithTag("", serviceName, conf)
 }
 
 func GrpcUrlWithTag(tag string, serviceName string, conf *myconf.Config) string {
+	return GrpcUrlWithTagByConfig(tag, serviceName, &ClientConfig{
+		Address:        conf.GetString("addr"),
+		Token:          conf.GetString("token"),
+		Username:       conf.GetString("username"),
+		Password:       conf.GetString("password"),
+		AutoGrpcPrefix: true,
+	})
+}
+
+func GrpcUrlWithTagByConfig(tag string, serviceName string, conf *ClientConfig) string {
+	if conf.AutoGrpcPrefix {
+		serviceName = mygrpc.ServicePrefix + serviceName
+	}
+
 	u := &url.URL{
 		Scheme: schemeName,
-		Host:   conf.GetString("addr"),
+		Host:   conf.Address,
 		Path:   serviceName,
 	}
 
@@ -195,44 +285,19 @@ func GrpcUrlWithTag(tag string, serviceName string, conf *myconf.Config) string
 	query := u.Query()
 	query.Set("healthy", "true")
 
-	if v := conf.GetString("token"); v != "" {
-		query.Set("token", v)
+	if conf.Token != "" {
+		query.Set("token", conf.Token)
 	}
 
 	if tag != "" {
 		query.Set("tag", tag)
 	}
 
-	username := conf.GetString("username")
-	password := conf.GetString("password")
-
-	if username != "" && password != "" {
-		u.User = url.UserPassword(username, password)
+	if conf.Username != "" && conf.Password != "" {
+		u.User = url.UserPassword(conf.Username, conf.Password)
 	}
 
 	u.RawQuery = query.Encode()
 
 	return u.String()
 }
-
-type watchLogger struct {
-	hclog.Logger
-	isWatchErr bool
-}
-
-func newWatchLogger() *watchLogger {
-	return &watchLogger{Logger: hclog.New(&hclog.LoggerOptions{
-		Name:   "watch",
-		Output: os.Stdout,
-	})}
-}
-
-func (l *watchLogger) Error(msg string, args ...interface{}) {
-	l.isWatchErr = true
-	log.Printf("is watch err: %s", msg)
-	l.Logger.Error(msg, args...)
-}
-
-func (l *watchLogger) Named(name string) hclog.Logger {
-	return l
-}

myregistry/consul/consul_test.go

@@ -0,0 +1,7 @@
+package consul
+
+import "testing"
+
+func TestWatch(t *testing.T) {
+
+}

myregistry/consul/resolver.go

@@ -0,0 +1,132 @@
+package consul
+
+import (
+	"fmt"
+	"github.com/hashicorp/consul/api"
+	"github.com/pkg/errors"
+	"google.golang.org/grpc/resolver"
+	"sort"
+	"strings"
+	"time"
+)
+
+const schemeName = "consul"
+
+func init() {
+	resolver.Register(&builder{})
+}
+
+var _ resolver.Builder = (*builder)(nil)
+
+type builder struct{}
+
+func (b *builder) Build(url resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
+	tgt, err := parseURL(url.URL.String())
+	if err != nil {
+		return nil, errors.Wrap(err, "Wrong consul URL")
+	}
+	client, err := NewClient(&ClientConfig{
+		Address:  tgt.Addr,
+		Token:    tgt.Token,
+		Username: tgt.User,
+		Password: tgt.Password,
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "Couldn't connect to the Consul API")
+	}
+
+	rl := &consulResolver{
+		client: client,
+		tgt:    tgt,
+		cc:     cc,
+	}
+
+	go rl.watchService()
+	return rl, nil
+}
+
+func (b *builder) Scheme() string {
+	return schemeName
+}
+
+var _ resolver.Resolver = (*consulResolver)(nil)
+
+type consulResolver struct {
+	client *api.Client
+	tgt    *target
+	cc     resolver.ClientConn
+}
+
+func (c *consulResolver) watchService() {
+	var (
+		lastIndex uint64
+		isFirst   = true
+	)
+
+	for {
+		if isFirst == false {
+			time.Sleep(time.Second * 2)
+		}
+		isFirst = false
+
+		endpoints, meta, err := c.client.Health().Service(c.tgt.Service, c.tgt.Tag,
+			true,
+			&api.QueryOptions{
+				WaitIndex:         lastIndex,
+				WaitTime:          c.tgt.Wait,
+				Datacenter:        c.tgt.Dc,
+				AllowStale:        c.tgt.AllowStale,
+				RequireConsistent: c.tgt.RequireConsistent,
+			})
+		if err != nil {
+			defaultLog.Errorf("watch service err: %s", err)
+			continue
+		}
+		lastIndex = meta.LastIndex
+
+		addrs := make([]string, 0, len(endpoints))
+
+		state := resolver.State{
+			Addresses: make([]resolver.Address, 0, len(endpoints)),
+		}
+		for _, endpoint := range endpoints {
+			tmp := resolver.Address{
+				Addr: endpoint.Service.Address,
+			}
+
+			if tmp.Addr == "" {
+				continue
+			}
+
+			tmp.Addr = fmt.Sprintf("%s:%d", tmp.Addr, endpoint.Service.Port)
+
+			state.Addresses = append(state.Addresses, tmp)
+			addrs = append(addrs, tmp.Addr)
+		}
+
+		if len(state.Addresses) == 0 {
+			defaultLog.Warnf("%s services num == 0", c.tgt.String())
+		}
+
+		sort.SliceStable(state.Addresses, func(i, j int) bool {
+			return state.Addresses[i].Addr < state.Addresses[j].Addr
+		})
+
+		err = c.cc.UpdateState(state)
+		if err != nil {
+			defaultLog.Errorf("%s update service state err: %s", err, c.tgt.String())
+		} else {
+			defaultLog.Infof("%s update service num:%d (%s)", c.tgt.String(), len(addrs), strings.Join(addrs, ", "))
+
+		}
+
+	}
+
+}
+
+func (c *consulResolver) ResolveNow(options resolver.ResolveNowOptions) {
+}
+
+func (c *consulResolver) Close() {
+
+}

myregistry/consul/target.go

@@ -29,30 +29,31 @@ type target struct {
 	Dc                string        `form:"dc"`
 	AllowStale        bool          `form:"allow-stale"`
 	RequireConsistent bool          `form:"require-consistent"`
-	// TODO(mbobakov): custom parameters for the http-transport
-	// TODO(mbobakov): custom parameters for the TLS subsystem
 }
 
 func (t *target) String() string {
-	return fmt.Sprintf("service='%s' healthy='%t' tag='%s'", t.Service, t.Healthy, t.Tag)
+	str := t.Service
+	if t.Tag != "" {
+		str = fmt.Sprintf("%s (tag:%s)", str, t.Tag)
+	}
+
+	return str
+
 }
 
-//  parseURL with parameters
-// see README.md for the actual format
-// URL schema will stay stable in the future for backward compatibility
-func parseURL(u string) (target, error) {
+func parseURL(u string) (*target, error) {
 	rawURL, err := url.Parse(u)
 	if err != nil {
-		return target{}, errors.Wrap(err, "Malformed URL")
+		return nil, errors.Wrap(err, "Malformed URL")
 	}
 
 	if rawURL.Scheme != schemeName ||
 		len(rawURL.Host) == 0 || len(strings.TrimLeft(rawURL.Path, "/")) == 0 {
-		return target{},
+		return nil,
 			errors.Errorf("Malformed URL('%s'). Must be in the next format: 'consul://[user:passwd]@host/service?param=value'", u)
 	}
 
-	var tgt target
+	tgt := &target{}
 	tgt.User = rawURL.User.Username()
 	tgt.Password, _ = rawURL.User.Password()
 	tgt.Addr = rawURL.Host
@@ -64,7 +65,7 @@ func parseURL(u string) (target, error) {
 
 	err = decoder.Decode(&tgt, rawURL.Query())
 	if err != nil {
-		return target{}, errors.Wrap(err, "Malformed URL parameters")
+		return nil, errors.Wrap(err, "Malformed URL parameters")
 	}
 	if len(tgt.Near) == 0 {
 		tgt.Near = "_agent"