2018-12-06 18:19:05 +00:00
|
|
|
// Package Gossip provides a gossip registry based on hashicorp/memberlist
|
2018-12-04 16:41:40 +00:00
|
|
|
package gossip
|
|
|
|
|
|
|
|
import (
|
2018-12-19 00:57:16 +03:00
|
|
|
"context"
|
2018-12-04 16:41:40 +00:00
|
|
|
"encoding/json"
|
2018-12-04 17:33:11 +00:00
|
|
|
"io/ioutil"
|
2018-12-19 00:57:16 +03:00
|
|
|
"net"
|
2018-12-04 16:41:40 +00:00
|
|
|
"os"
|
2019-01-30 15:39:57 +03:00
|
|
|
"os/signal"
|
2018-12-19 00:57:16 +03:00
|
|
|
"strconv"
|
2018-12-04 16:41:40 +00:00
|
|
|
"strings"
|
|
|
|
"sync"
|
2019-01-30 15:39:57 +03:00
|
|
|
"syscall"
|
2018-12-04 16:41:40 +00:00
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/golang/protobuf/proto"
|
|
|
|
"github.com/google/uuid"
|
|
|
|
"github.com/hashicorp/memberlist"
|
2018-12-19 00:57:16 +03:00
|
|
|
log "github.com/micro/go-log"
|
2018-12-04 16:41:40 +00:00
|
|
|
"github.com/micro/go-micro/registry"
|
|
|
|
pb "github.com/micro/go-micro/registry/gossip/proto"
|
2018-12-06 18:19:05 +00:00
|
|
|
"github.com/mitchellh/hashstructure"
|
2018-12-04 16:41:40 +00:00
|
|
|
)
|
|
|
|
|
2019-01-30 15:39:57 +03:00
|
|
|
// use registry.Result int32 values after it switches from string to int32 types
|
|
|
|
// type actionType int32
|
|
|
|
// type updateType int32
|
|
|
|
|
|
|
|
const (
|
|
|
|
actionTypeInvalid int32 = iota
|
|
|
|
actionTypeCreate
|
|
|
|
actionTypeDelete
|
|
|
|
actionTypeUpdate
|
|
|
|
actionTypeSync
|
|
|
|
)
|
|
|
|
|
|
|
|
func actionTypeString(t int32) string {
|
|
|
|
switch t {
|
|
|
|
case actionTypeCreate:
|
|
|
|
return "create"
|
|
|
|
case actionTypeDelete:
|
|
|
|
return "delete"
|
|
|
|
case actionTypeUpdate:
|
|
|
|
return "update"
|
|
|
|
case actionTypeSync:
|
|
|
|
return "sync"
|
|
|
|
}
|
|
|
|
return "invalid"
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
const (
|
2019-01-30 15:39:57 +03:00
|
|
|
updateTypeInvalid int32 = iota
|
|
|
|
updateTypeService
|
2018-12-06 18:19:05 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
type broadcast struct {
|
|
|
|
update *pb.Update
|
|
|
|
notify chan<- struct{}
|
|
|
|
}
|
|
|
|
|
|
|
|
type delegate struct {
|
|
|
|
queue *memberlist.TransmitLimitedQueue
|
|
|
|
updates chan *update
|
|
|
|
}
|
|
|
|
|
2018-12-04 16:41:40 +00:00
|
|
|
type gossipRegistry struct {
|
2018-12-06 18:19:05 +00:00
|
|
|
queue *memberlist.TransmitLimitedQueue
|
|
|
|
updates chan *update
|
|
|
|
options registry.Options
|
|
|
|
member *memberlist.Memberlist
|
|
|
|
interval time.Duration
|
2018-12-04 16:41:40 +00:00
|
|
|
|
|
|
|
sync.RWMutex
|
|
|
|
services map[string][]*registry.Service
|
2018-12-06 18:19:05 +00:00
|
|
|
|
|
|
|
s sync.RWMutex
|
|
|
|
watchers map[string]chan *registry.Result
|
|
|
|
}
|
|
|
|
|
|
|
|
type update struct {
|
|
|
|
Update *pb.Update
|
|
|
|
Service *registry.Service
|
|
|
|
sync chan *registry.Service
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
var (
|
2018-12-06 18:19:05 +00:00
|
|
|
// You should change this if using secure
|
2018-12-19 00:57:16 +03:00
|
|
|
DefaultSecret = []byte("micro-gossip-key") // exactly 16 bytes
|
2018-12-06 18:19:05 +00:00
|
|
|
ExpiryTick = time.Second * 5
|
2018-12-04 16:41:40 +00:00
|
|
|
)
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
func configure(g *gossipRegistry, opts ...registry.Option) error {
|
|
|
|
// loop through address list and get valid entries
|
|
|
|
addrs := func(curAddrs []string) []string {
|
|
|
|
var newAddrs []string
|
|
|
|
for _, addr := range curAddrs {
|
|
|
|
if trimAddr := strings.TrimSpace(addr); len(trimAddr) > 0 {
|
|
|
|
newAddrs = append(newAddrs, trimAddr)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return newAddrs
|
|
|
|
}
|
2018-12-04 16:41:40 +00:00
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
// current address list
|
|
|
|
curAddrs := addrs(g.options.Addrs)
|
|
|
|
|
|
|
|
// parse options
|
|
|
|
for _, o := range opts {
|
|
|
|
o(&g.options)
|
|
|
|
}
|
|
|
|
|
|
|
|
// new address list
|
|
|
|
newAddrs := addrs(g.options.Addrs)
|
|
|
|
|
|
|
|
// no new nodes and existing member. no configure
|
|
|
|
if (len(newAddrs) == len(curAddrs)) && g.member != nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// shutdown old member
|
|
|
|
if g.member != nil {
|
2019-01-30 15:39:57 +03:00
|
|
|
g.Stop()
|
2018-12-06 18:19:05 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// replace addresses
|
|
|
|
curAddrs = newAddrs
|
|
|
|
|
|
|
|
// create a new default config
|
|
|
|
c := memberlist.DefaultLocalConfig()
|
|
|
|
|
2019-01-30 15:39:57 +03:00
|
|
|
// log to dev null
|
|
|
|
c.LogOutput = ioutil.Discard
|
|
|
|
|
2018-12-19 00:57:16 +03:00
|
|
|
if optConfig, ok := g.options.Context.Value(contextConfig{}).(*memberlist.Config); ok && optConfig != nil {
|
|
|
|
c = optConfig
|
|
|
|
}
|
|
|
|
|
|
|
|
if hostport, ok := g.options.Context.Value(contextAddress{}).(string); ok {
|
|
|
|
host, port, err := net.SplitHostPort(hostport)
|
|
|
|
if err == nil {
|
|
|
|
pn, err := strconv.Atoi(port)
|
|
|
|
if err == nil {
|
|
|
|
c.BindPort = pn
|
|
|
|
}
|
|
|
|
c.BindAddr = host
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// set bind to random port
|
|
|
|
c.BindPort = 0
|
|
|
|
}
|
|
|
|
|
|
|
|
if hostport, ok := g.options.Context.Value(contextAdvertise{}).(string); ok {
|
|
|
|
host, port, err := net.SplitHostPort(hostport)
|
|
|
|
if err == nil {
|
|
|
|
pn, err := strconv.Atoi(port)
|
|
|
|
if err == nil {
|
|
|
|
c.AdvertisePort = pn
|
|
|
|
}
|
|
|
|
c.AdvertiseAddr = host
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// machine hostname
|
|
|
|
hostname, _ := os.Hostname()
|
2018-12-06 18:19:05 +00:00
|
|
|
|
|
|
|
// set the name
|
|
|
|
c.Name = strings.Join([]string{"micro", hostname, uuid.New().String()}, "-")
|
|
|
|
|
|
|
|
// set a secret key if secure
|
|
|
|
if g.options.Secure {
|
|
|
|
k, ok := g.options.Context.Value(contextSecretKey{}).([]byte)
|
|
|
|
if !ok {
|
|
|
|
// use the default secret
|
|
|
|
k = DefaultSecret
|
|
|
|
}
|
|
|
|
c.SecretKey = k
|
|
|
|
}
|
|
|
|
|
2019-01-30 15:39:57 +03:00
|
|
|
// create a queue
|
|
|
|
queue := &memberlist.TransmitLimitedQueue{
|
|
|
|
NumNodes: func() int {
|
|
|
|
return len(curAddrs)
|
|
|
|
},
|
|
|
|
RetransmitMult: 3,
|
|
|
|
}
|
|
|
|
|
|
|
|
// set the delegate
|
|
|
|
c.Delegate = &delegate{
|
|
|
|
updates: g.updates,
|
|
|
|
queue: queue,
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
// create the memberlist
|
|
|
|
m, err := memberlist.Create(c)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
// join the memberlist
|
|
|
|
if len(curAddrs) > 0 {
|
|
|
|
_, err := m.Join(curAddrs)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// set internals
|
|
|
|
g.queue = queue
|
|
|
|
g.member = m
|
|
|
|
g.interval = c.GossipInterval
|
|
|
|
|
|
|
|
log.Logf("Registry Listening on %s", m.LocalNode().Address())
|
|
|
|
return nil
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
2019-01-30 15:39:57 +03:00
|
|
|
func (*broadcast) UniqueBroadcast() {}
|
2018-12-04 16:41:40 +00:00
|
|
|
|
2019-01-30 15:39:57 +03:00
|
|
|
func (b *broadcast) Invalidates(other memberlist.Broadcast) bool {
|
|
|
|
return false
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (b *broadcast) Message() []byte {
|
|
|
|
up, err := proto.Marshal(b.update)
|
|
|
|
if err != nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
return up
|
|
|
|
}
|
|
|
|
|
|
|
|
func (b *broadcast) Finished() {
|
|
|
|
if b.notify != nil {
|
|
|
|
close(b.notify)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *delegate) NodeMeta(limit int) []byte {
|
|
|
|
return []byte{}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *delegate) NotifyMsg(b []byte) {
|
|
|
|
if len(b) == 0 {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
go func() {
|
|
|
|
up := new(pb.Update)
|
|
|
|
if err := proto.Unmarshal(b, up); err != nil {
|
2018-12-04 16:41:40 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
// only process service action
|
2019-01-30 15:39:57 +03:00
|
|
|
if up.Type != updateTypeService {
|
2018-12-06 18:19:05 +00:00
|
|
|
return
|
|
|
|
}
|
2018-12-04 16:41:40 +00:00
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
var service *registry.Service
|
2018-12-04 16:41:40 +00:00
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
switch up.Metadata["Content-Type"] {
|
|
|
|
case "application/json":
|
|
|
|
if err := json.Unmarshal(up.Data, &service); err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
// no other content type
|
|
|
|
default:
|
2018-12-04 16:41:40 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
// send update
|
|
|
|
d.updates <- &update{
|
|
|
|
Update: up,
|
|
|
|
Service: service,
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
2018-12-06 18:19:05 +00:00
|
|
|
}()
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
|
|
|
|
return d.queue.GetBroadcasts(overhead, limit)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *delegate) LocalState(join bool) []byte {
|
2018-12-06 18:19:05 +00:00
|
|
|
if !join {
|
|
|
|
return []byte{}
|
|
|
|
}
|
|
|
|
|
|
|
|
syncCh := make(chan *registry.Service, 1)
|
|
|
|
services := map[string][]*registry.Service{}
|
|
|
|
|
|
|
|
d.updates <- &update{
|
|
|
|
Update: &pb.Update{
|
2019-01-30 15:39:57 +03:00
|
|
|
Action: actionTypeSync,
|
2018-12-06 18:19:05 +00:00
|
|
|
},
|
|
|
|
sync: syncCh,
|
|
|
|
}
|
|
|
|
|
|
|
|
for srv := range syncCh {
|
|
|
|
services[srv.Name] = append(services[srv.Name], srv)
|
|
|
|
}
|
|
|
|
|
|
|
|
b, _ := json.Marshal(services)
|
2018-12-04 16:41:40 +00:00
|
|
|
return b
|
|
|
|
}
|
|
|
|
|
|
|
|
func (d *delegate) MergeRemoteState(buf []byte, join bool) {
|
|
|
|
if len(buf) == 0 {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
if !join {
|
|
|
|
return
|
|
|
|
}
|
2018-12-06 18:19:05 +00:00
|
|
|
|
2018-12-04 16:41:40 +00:00
|
|
|
var services map[string][]*registry.Service
|
|
|
|
if err := json.Unmarshal(buf, &services); err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
for _, service := range services {
|
|
|
|
for _, srv := range service {
|
|
|
|
d.updates <- &update{
|
2019-01-30 15:39:57 +03:00
|
|
|
Update: &pb.Update{Action: actionTypeCreate},
|
2018-12-06 18:19:05 +00:00
|
|
|
Service: srv,
|
|
|
|
sync: nil,
|
|
|
|
}
|
|
|
|
}
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
func (g *gossipRegistry) publish(action string, services []*registry.Service) {
|
|
|
|
g.s.RLock()
|
|
|
|
for _, sub := range g.watchers {
|
|
|
|
go func(sub chan *registry.Result) {
|
|
|
|
for _, service := range services {
|
|
|
|
sub <- ®istry.Result{Action: action, Service: service}
|
|
|
|
}
|
|
|
|
}(sub)
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
2018-12-06 18:19:05 +00:00
|
|
|
g.s.RUnlock()
|
|
|
|
}
|
2018-12-04 16:41:40 +00:00
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
func (g *gossipRegistry) subscribe() (chan *registry.Result, chan bool) {
|
|
|
|
next := make(chan *registry.Result, 10)
|
|
|
|
exit := make(chan bool)
|
|
|
|
|
|
|
|
id := uuid.New().String()
|
|
|
|
|
|
|
|
g.s.Lock()
|
|
|
|
g.watchers[id] = next
|
|
|
|
g.s.Unlock()
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
<-exit
|
|
|
|
g.s.Lock()
|
|
|
|
delete(g.watchers, id)
|
|
|
|
close(next)
|
|
|
|
g.s.Unlock()
|
|
|
|
}()
|
|
|
|
|
|
|
|
return next, exit
|
|
|
|
}
|
|
|
|
|
2019-01-30 15:39:57 +03:00
|
|
|
func (g *gossipRegistry) wait() {
|
|
|
|
ctx := g.options.Context
|
|
|
|
|
|
|
|
if c, ok := ctx.Value(contextContext{}).(context.Context); ok && c != nil {
|
|
|
|
ctx = c
|
|
|
|
}
|
|
|
|
|
|
|
|
ch := make(chan os.Signal, 1)
|
|
|
|
signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT)
|
|
|
|
|
|
|
|
select {
|
|
|
|
// wait on kill signal
|
|
|
|
case <-ch:
|
|
|
|
// wait on context cancel
|
|
|
|
case <-ctx.Done():
|
|
|
|
}
|
|
|
|
|
|
|
|
g.Stop()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *gossipRegistry) Stop() {
|
|
|
|
g.member.Leave(g.interval * 2)
|
|
|
|
g.member.Shutdown()
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
func (g *gossipRegistry) run() {
|
|
|
|
var mtx sync.Mutex
|
|
|
|
updates := map[uint64]*update{}
|
|
|
|
|
|
|
|
// expiry loop
|
|
|
|
go func() {
|
|
|
|
t := time.NewTicker(ExpiryTick)
|
|
|
|
defer t.Stop()
|
|
|
|
|
|
|
|
for _ = range t.C {
|
|
|
|
now := uint64(time.Now().UnixNano())
|
|
|
|
|
|
|
|
mtx.Lock()
|
|
|
|
|
|
|
|
// process all the updates
|
|
|
|
for k, v := range updates {
|
|
|
|
// check if expiry time has passed
|
2019-01-30 15:39:57 +03:00
|
|
|
if d := (v.Update.Expires); d < now {
|
2018-12-06 18:19:05 +00:00
|
|
|
// delete from records
|
|
|
|
delete(updates, k)
|
|
|
|
// set to delete
|
2019-01-30 15:39:57 +03:00
|
|
|
v.Update.Action = actionTypeDelete
|
2018-12-06 18:19:05 +00:00
|
|
|
// fire a new update
|
|
|
|
g.updates <- v
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
mtx.Unlock()
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
// process the updates
|
|
|
|
for u := range g.updates {
|
|
|
|
switch u.Update.Action {
|
2019-01-30 15:39:57 +03:00
|
|
|
case actionTypeCreate:
|
2018-12-06 18:19:05 +00:00
|
|
|
g.Lock()
|
|
|
|
if service, ok := g.services[u.Service.Name]; !ok {
|
|
|
|
g.services[u.Service.Name] = []*registry.Service{u.Service}
|
|
|
|
|
|
|
|
} else {
|
|
|
|
g.services[u.Service.Name] = addServices(service, []*registry.Service{u.Service})
|
|
|
|
}
|
|
|
|
g.Unlock()
|
|
|
|
|
|
|
|
// publish update to watchers
|
2019-01-30 15:39:57 +03:00
|
|
|
go g.publish(actionTypeString(actionTypeCreate), []*registry.Service{u.Service})
|
2018-12-06 18:19:05 +00:00
|
|
|
|
|
|
|
// we need to expire the node at some point in the future
|
|
|
|
if u.Update.Expires > 0 {
|
|
|
|
// create a hash of this service
|
|
|
|
if hash, err := hashstructure.Hash(u.Service, nil); err == nil {
|
|
|
|
mtx.Lock()
|
|
|
|
updates[hash] = u
|
|
|
|
mtx.Unlock()
|
|
|
|
}
|
|
|
|
}
|
2019-01-30 15:39:57 +03:00
|
|
|
case actionTypeDelete:
|
2018-12-06 18:19:05 +00:00
|
|
|
g.Lock()
|
|
|
|
if service, ok := g.services[u.Service.Name]; ok {
|
|
|
|
if services := delServices(service, []*registry.Service{u.Service}); len(services) == 0 {
|
|
|
|
delete(g.services, u.Service.Name)
|
|
|
|
} else {
|
|
|
|
g.services[u.Service.Name] = services
|
|
|
|
}
|
|
|
|
}
|
|
|
|
g.Unlock()
|
|
|
|
|
|
|
|
// publish update to watchers
|
2019-01-30 15:39:57 +03:00
|
|
|
go g.publish(actionTypeString(actionTypeDelete), []*registry.Service{u.Service})
|
2018-12-06 18:19:05 +00:00
|
|
|
|
|
|
|
// delete from expiry checks
|
|
|
|
if hash, err := hashstructure.Hash(u.Service, nil); err == nil {
|
|
|
|
mtx.Lock()
|
|
|
|
delete(updates, hash)
|
|
|
|
mtx.Unlock()
|
|
|
|
}
|
2019-01-30 15:39:57 +03:00
|
|
|
case actionTypeSync:
|
2018-12-06 18:19:05 +00:00
|
|
|
// no sync channel provided
|
|
|
|
if u.sync == nil {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
g.RLock()
|
|
|
|
|
|
|
|
// push all services through the sync chan
|
|
|
|
for _, service := range g.services {
|
|
|
|
for _, srv := range service {
|
|
|
|
u.sync <- srv
|
|
|
|
}
|
|
|
|
|
|
|
|
// publish to watchers
|
2019-01-30 15:39:57 +03:00
|
|
|
go g.publish(actionTypeString(actionTypeCreate), service)
|
2018-12-06 18:19:05 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
g.RUnlock()
|
|
|
|
|
|
|
|
// close the sync chan
|
|
|
|
close(u.sync)
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
}
|
2018-12-06 18:19:05 +00:00
|
|
|
}
|
2018-12-04 16:41:40 +00:00
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
func (g *gossipRegistry) Init(opts ...registry.Option) error {
|
|
|
|
return configure(g, opts...)
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (g *gossipRegistry) Options() registry.Options {
|
2018-12-06 18:19:05 +00:00
|
|
|
return g.options
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (g *gossipRegistry) Register(s *registry.Service, opts ...registry.RegisterOption) error {
|
|
|
|
b, err := json.Marshal(s)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
g.Lock()
|
2018-12-06 18:19:05 +00:00
|
|
|
if service, ok := g.services[s.Name]; !ok {
|
|
|
|
g.services[s.Name] = []*registry.Service{s}
|
|
|
|
} else {
|
|
|
|
g.services[s.Name] = addServices(service, []*registry.Service{s})
|
|
|
|
}
|
2018-12-04 16:41:40 +00:00
|
|
|
g.Unlock()
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
var options registry.RegisterOptions
|
|
|
|
for _, o := range opts {
|
|
|
|
o(&options)
|
|
|
|
}
|
|
|
|
|
2018-12-04 16:41:40 +00:00
|
|
|
up := &pb.Update{
|
2019-01-30 15:39:57 +03:00
|
|
|
Expires: uint64(time.Now().Add(options.TTL).UnixNano()),
|
|
|
|
Action: actionTypeCreate,
|
|
|
|
Type: updateTypeService,
|
2018-12-04 16:41:40 +00:00
|
|
|
Metadata: map[string]string{
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
},
|
|
|
|
Data: b,
|
|
|
|
}
|
|
|
|
|
|
|
|
g.queue.QueueBroadcast(&broadcast{
|
|
|
|
update: up,
|
|
|
|
notify: nil,
|
|
|
|
})
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
// wait
|
|
|
|
<-time.After(g.interval * 2)
|
|
|
|
|
2018-12-04 16:41:40 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *gossipRegistry) Deregister(s *registry.Service) error {
|
|
|
|
b, err := json.Marshal(s)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
g.Lock()
|
2018-12-06 18:19:05 +00:00
|
|
|
if service, ok := g.services[s.Name]; ok {
|
|
|
|
if services := delServices(service, []*registry.Service{s}); len(services) == 0 {
|
|
|
|
delete(g.services, s.Name)
|
|
|
|
} else {
|
|
|
|
g.services[s.Name] = services
|
|
|
|
}
|
|
|
|
}
|
2018-12-04 16:41:40 +00:00
|
|
|
g.Unlock()
|
|
|
|
|
|
|
|
up := &pb.Update{
|
2019-01-30 15:39:57 +03:00
|
|
|
Action: actionTypeDelete,
|
|
|
|
Type: updateTypeService,
|
2018-12-04 16:41:40 +00:00
|
|
|
Metadata: map[string]string{
|
|
|
|
"Content-Type": "application/json",
|
|
|
|
},
|
|
|
|
Data: b,
|
|
|
|
}
|
|
|
|
|
|
|
|
g.queue.QueueBroadcast(&broadcast{
|
|
|
|
update: up,
|
|
|
|
notify: nil,
|
|
|
|
})
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
// wait
|
|
|
|
<-time.After(g.interval * 2)
|
|
|
|
|
2018-12-04 16:41:40 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *gossipRegistry) GetService(name string) ([]*registry.Service, error) {
|
|
|
|
g.RLock()
|
2018-12-06 18:19:05 +00:00
|
|
|
service, ok := g.services[name]
|
2018-12-04 16:41:40 +00:00
|
|
|
g.RUnlock()
|
2018-12-06 18:19:05 +00:00
|
|
|
if !ok {
|
|
|
|
return nil, registry.ErrNotFound
|
|
|
|
}
|
|
|
|
return service, nil
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (g *gossipRegistry) ListServices() ([]*registry.Service, error) {
|
|
|
|
var services []*registry.Service
|
|
|
|
g.RLock()
|
2018-12-06 18:19:05 +00:00
|
|
|
for _, service := range g.services {
|
|
|
|
services = append(services, service...)
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
g.RUnlock()
|
|
|
|
return services, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (g *gossipRegistry) Watch(opts ...registry.WatchOption) (registry.Watcher, error) {
|
2018-12-06 18:19:05 +00:00
|
|
|
n, e := g.subscribe()
|
|
|
|
return newGossipWatcher(n, e, opts...)
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func (g *gossipRegistry) String() string {
|
|
|
|
return "gossip"
|
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
func NewRegistry(opts ...registry.Option) registry.Registry {
|
|
|
|
gossip := &gossipRegistry{
|
2018-12-19 09:27:53 +00:00
|
|
|
options: registry.Options{
|
|
|
|
Context: context.Background(),
|
|
|
|
},
|
2018-12-06 18:19:05 +00:00
|
|
|
updates: make(chan *update, 100),
|
|
|
|
services: make(map[string][]*registry.Service),
|
|
|
|
watchers: make(map[string]chan *registry.Result),
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
2018-12-19 19:04:44 +00:00
|
|
|
// run the updater
|
|
|
|
go gossip.run()
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
// configure the gossiper
|
|
|
|
if err := configure(gossip, opts...); err != nil {
|
2018-12-19 00:57:16 +03:00
|
|
|
log.Fatalf("Error configuring registry: %v", err)
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
// wait for setup
|
|
|
|
<-time.After(gossip.interval * 2)
|
2018-12-04 16:41:40 +00:00
|
|
|
|
2019-01-30 15:39:57 +03:00
|
|
|
go gossip.wait()
|
|
|
|
|
2018-12-06 18:19:05 +00:00
|
|
|
return gossip
|
2018-12-04 16:41:40 +00:00
|
|
|
}
|