实战etcd的服务发现

// Builder creates a resolver that will be used to watch name resolution updates.
type Builder interface {
	// Build creates a new resolver for the given target.
	//
	// gRPC dial calls Build synchronously, and fails if the returned error is
	// not nil.
	Build(target Target, cc ClientConn, opts BuildOptions) (Resolver, error)
	// Scheme returns the scheme supported by this resolver.
	// Scheme is defined at https://github.com/grpc/grpc/blob/master/doc/naming.md.
	Scheme() string
}

// Resolver watches for the updates on the specified target.
// Updates include address updates and service config updates.
type Resolver interface {
	// ResolveNow will be called by gRPC to try to resolve the target name
	// again. It's just a hint, resolver can ignore this if it's not necessary.
	//
	// It could be called multiple times concurrently.
	ResolveNow(ResolveNowOptions)
	// Close closes the resolver.
	Close()
}

package etcd

import (
	"fmt"

	"go.etcd.io/etcd/clientv3"
	"google.golang.org/grpc/resolver"
)

type Builder struct {
	Client *clientv3.Client
}

func (b *Builder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	prefix := fmt.Sprintf("/%s/", target.Endpoint)

	r := &Resolver{
		Client: b.Client,
		cc:     cc,
		prefix: prefix,
	}

	go r.watcher()
	r.ResolveNow(resolver.ResolveNowOptions{})
	return r, nil
}

func (b *Builder) Scheme() string {
	return "etcd"
}

package etcd

import (
	"context"
	"sync"

	"go.etcd.io/etcd/clientv3"
	"go.etcd.io/etcd/mvcc/mvccpb"
	"google.golang.org/grpc/resolver"
)

type Resolver struct {
	sync.RWMutex
	Client    *clientv3.Client
	cc        resolver.ClientConn
	prefix    string
	addresses map[string]resolver.Address
}

func (r *Resolver) ResolveNow(resolver.ResolveNowOptions) {
	// todo
}

func (r *Resolver) Close() {
	// todo
}

func (r *Resolver) watcher() {
	r.addresses = make(map[string]resolver.Address)

	response, err := r.Client.Get(context.Background(), r.prefix, clientv3.WithPrefix())

	if err == nil {
		for _, kv := range response.Kvs {
			r.setAddress(string(kv.Key), string(kv.Value))
		}

		r.cc.UpdateState(resolver.State{
			Addresses: r.getAddresses(),
		})
	}

	watch := r.Client.Watch(context.Background(), r.prefix, clientv3.WithPrefix())

	for response := range watch {
		for _, event := range response.Events {
			switch event.Type {
			case mvccpb.PUT:
				r.setAddress(string(event.Kv.Key), string(event.Kv.Value))
			case mvccpb.DELETE:
				r.delAddress(string(event.Kv.Key))
			}
		}

		r.cc.UpdateState(resolver.State{
			Addresses: r.getAddresses(),
		})
	}
}

func (r *Resolver) setAddress(key, address string) {
	r.Lock()
	defer r.Unlock()
	r.addresses[key] = resolver.Address{Addr: string(address)}
}

func (r *Resolver) delAddress(key string) {
	r.Lock()
	defer r.Unlock()
	delete(r.addresses, key)
}

func (r *Resolver) getAddresses() []resolver.Address {
	addresses := make([]resolver.Address, 0, len(r.addresses))

	for _, address := range r.addresses {
		addresses = append(addresses, address)
	}

	return addresses
}

func main() {
	host := viper.GetString("server.host")
	port := viper.GetString("server.port")
	listener, err := net.Listen("tcp", net.JoinHostPort(host, port))

	if err != nil {
		log.Fatalln(err)
	}

	server := grpc.NewServer()
	reflection.Register(server)
	pb.RegisterFooServer(server, &foo.Server{})
	close, err := register("foo", 10)

	if err != nil {
		log.Fatalln(err)
	}

	go func() {
		if err := server.Serve(listener); err != nil {
			log.Fatalln(err)
		}
	}()

	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit
	close()
}

func register(service string, ttl int64) (func(), error) {
	port := viper.GetString("server.port")
	client, err := etcdClient()

	if err != nil {
		return nil, err
	}

	ctx := context.Background()
	lease, err := client.Grant(ctx, ttl)

	if err != nil {
		return nil, err
	}

	ip, err := localIP()

	if err != nil {
		return nil, err
	}

	key := fmt.Sprintf("/%s/%s:%s", service, ip, port)
	value := fmt.Sprintf("%s:%s", ip, port)
	_, err = client.Put(ctx, key, value, clientv3.WithLease(lease.ID))

	if err != nil {
		return nil, err
	}

	keepAlive, err := client.KeepAlive(ctx, lease.ID)

	if err != nil {
		return nil, err
	}

	go func() {
		for {
			<-keepAlive
		}
	}()

	close := func() {
		_, _ = client.Revoke(ctx, lease.ID)
	}

	return close, nil
}

func localIP() (string, error) {
	addrs, err := net.InterfaceAddrs()

	if err != nil {
		return "", err
	}

	for _, addr := range addrs {
		if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
			if ipnet.IP.To4() != nil {
				return ipnet.IP.String(), nil
			}
		}
	}

	return "", errors.New("unable to determine local ip")
}

func main() {
	client, err := etcdClient()

	if err != nil {
		log.Fatalln(err)
	}

	builder := &etcd.Builder{
		Client: client,
	}

	resolver.Register(builder)
	ctx := context.Background()
	target := "etcd:///foo"

	cc, err := grpc.DialContext(
		ctx,
		target,
		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, roundrobin.Name)),
		grpc.WithInsecure(),
	)

	if err != nil {
		log.Fatalln(err)
	}

	defer cc.Close()

	// pb.NewFooClient(cc) ...
}

replace (
	github.com/golang/protobuf => github.com/golang/protobuf v1.3.2
	go.etcd.io/etcd => go.etcd.io/etcd v0.0.0-20191023171146-3cf2f69b5738
	google.golang.org/grpc => google.golang.org/grpc v1.26.0
)