Merge pull request #492 from yurishkuro/ys-discovery-via-push

Support push model for service discovery

examples/apigateway/main.go

@@ -84,18 +84,19 @@ func main() {
  tags = []string{}
  passingOnly = true
  endpoints = addsvc.Endpoints{}
+ instancer = consulsd.NewInstancer(client, logger, "addsvc", tags, passingOnly)
  )
  {
  factory := addsvcFactory(addsvc.MakeSumEndpoint, tracer, logger)
- subscriber := consulsd.NewSubscriber(client, factory, logger, "addsvc", tags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(*retryMax, *retryTimeout, balancer)
  endpoints.SumEndpoint = retry
  }
  {
  factory := addsvcFactory(addsvc.MakeConcatEndpoint, tracer, logger)
- subscriber := consulsd.NewSubscriber(client, factory, logger, "addsvc", tags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(*retryMax, *retryTimeout, balancer)
  endpoints.ConcatEndpoint = retry
  }
@@ -120,18 +121,19 @@ func main() {
  passingOnly = true
  uppercase endpoint.Endpoint
  count endpoint.Endpoint
+ instancer = consulsd.NewInstancer(client, logger, "stringsvc", tags, passingOnly)
  )
  {
  factory := stringsvcFactory(ctx, "GET", "/uppercase")
- subscriber := consulsd.NewSubscriber(client, factory, logger, "stringsvc", tags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(*retryMax, *retryTimeout, balancer)
  uppercase = retry
  }
  {
  factory := stringsvcFactory(ctx, "GET", "/count")
- subscriber := consulsd.NewSubscriber(client, factory, logger, "stringsvc", tags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(*retryMax, *retryTimeout, balancer)
  count = retry
  }

examples/profilesvc/client/client.go

@@ -40,68 +40,69 @@ func New(consulAddr string, logger log.Logger) (profilesvc.Service, error) {
 
  var (
  sdclient = consul.NewClient(apiclient)
+ instancer = consul.NewInstancer(sdclient, logger, consulService, consulTags, passingOnly)
  endpoints profilesvc.Endpoints
  )
  {
  factory := factoryFor(profilesvc.MakePostProfileEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.PostProfileEndpoint = retry
  }
  {
  factory := factoryFor(profilesvc.MakeGetProfileEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.GetProfileEndpoint = retry
  }
  {
  factory := factoryFor(profilesvc.MakePutProfileEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.PutProfileEndpoint = retry
  }
  {
  factory := factoryFor(profilesvc.MakePatchProfileEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.PatchProfileEndpoint = retry
  }
  {
  factory := factoryFor(profilesvc.MakeDeleteProfileEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.DeleteProfileEndpoint = retry
  }
  {
  factory := factoryFor(profilesvc.MakeGetAddressesEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.GetAddressesEndpoint = retry
  }
  {
  factory := factoryFor(profilesvc.MakeGetAddressEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.GetAddressEndpoint = retry
  }
  {
  factory := factoryFor(profilesvc.MakePostAddressEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.PostAddressEndpoint = retry
  }
  {
  factory := factoryFor(profilesvc.MakeDeleteAddressEndpoint)
- subscriber := consul.NewSubscriber(sdclient, factory, logger, consulService, consulTags, passingOnly)
- balancer := lb.NewRoundRobin(subscriber)
+ endpointer := sd.NewEndpointer(instancer, factory, logger)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(retryMax, retryTimeout, balancer)
  endpoints.DeleteAddressEndpoint = retry
  }

examples/stringsvc3/proxying.go

@@ -40,20 +40,20 @@ func proxyingMiddleware(ctx context.Context, instances string, logger log.Logger
  // discovery system.
  var (
  instanceList = split(instances)
- subscriber sd.FixedSubscriber
+ endpointer sd.FixedEndpointer
  )
  logger.Log("proxy_to", fmt.Sprint(instanceList))
  for _, instance := range instanceList {
  var e endpoint.Endpoint
  e = makeUppercaseProxy(ctx, instance)
  e = circuitbreaker.Gobreaker(gobreaker.NewCircuitBreaker(gobreaker.Settings{}))(e)
  e = ratelimit.NewTokenBucketLimiter(jujuratelimit.NewBucketWithRate(float64(qps), int64(qps)))(e)
- subscriber = append(subscriber, e)
+ endpointer = append(endpointer, e)
  }
 
  // Now, build a single, retrying, load-balancing endpoint out of all of
  // those individual endpoints.
- balancer := lb.NewRoundRobin(subscriber)
+ balancer := lb.NewRoundRobin(endpointer)
  retry := lb.Retry(maxAttempts, maxTime, balancer)
 
  // And finally, return the ServiceMiddleware, implemented by proxymw.

sd/cache/benchmark_test.go → sd/benchmark_test.go

@@ -1,4 +1,4 @@
-package cache
+package sd
 
 import (
  "io"
@@ -14,12 +14,12 @@ func BenchmarkEndpoints(b *testing.B) {
  cb = make(closer)
  cmap = map[string]io.Closer{"a": ca, "b": cb}
  factory = func(instance string) (endpoint.Endpoint, io.Closer, error) { return endpoint.Nop, cmap[instance], nil }
- c = New(factory, log.NewNopLogger())
+ c = newEndpointCache(factory, log.NewNopLogger(), endpointerOptions{})
  )
 
  b.ReportAllocs()
 
- c.Update([]string{"a", "b"})
+ c.Update(Event{Instances: []string{"a", "b"}})
 
  b.RunParallel(func(pb *testing.PB) {
  for pb.Next() {

sd/consul/doc.go

@@ -1,2 +1,2 @@
-// Package consul provides subscriber and registrar implementations for Consul.
+// Package consul provides Instancer and Registrar implementations for Consul.
 package consul