Codebase list golang-github-go-kit-kit / d0853ee
Merge pull request #504 from martinbaillie/master Add Eureka service discovery implementation Peter Bourgon authored 7 years ago GitHub committed 7 years ago
9 changed file(s) with 702 addition(s) and 0 deletion(s). Raw diff Collapse all Expand all
+1
-0
circle.yml less more
2323 ETCD_ADDR: http://localhost:2379
2424 CONSUL_ADDR: localhost:8500
2525 ZK_ADDR: localhost:2181
26 EUREKA_ADDR: http://localhost:8761/eureka
+6
-0
docker-compose-integration.yml less more
1313 image: zookeeper
1414 ports:
1515 - "2181:2181"
16 eureka:
17 image: springcloud/eureka
18 environment:
19 eureka.server.responseCacheUpdateIntervalMs: 1000
20 ports:
21 - "8761:8761"
+2
-0
sd/eureka/doc.go less more
0 // Package eureka provides subscriber and registrar implementations for Netflix OSS's Eureka
1 package eureka
+106
-0
sd/eureka/integration_test.go less more
0 // +build integration
1
2 package eureka
3
4 import (
5 "io"
6 "os"
7 "testing"
8 "time"
9
10 "github.com/hudl/fargo"
11
12 "github.com/go-kit/kit/endpoint"
13 "github.com/go-kit/kit/log"
14 )
15
16 // Package sd/eureka provides a wrapper around the Netflix Eureka service
17 // registry by way of the Fargo library. This test assumes the user has an
18 // instance of Eureka available at the address in the environment variable.
19 // Example `${EUREKA_ADDR}` format: http://localhost:8761/eureka
20 //
21 // NOTE: when starting a Eureka server for integration testing, ensure
22 // the response cache interval is reduced to one second. This can be
23 // achieved with the following Java argument:
24 // `-Deureka.server.responseCacheUpdateIntervalMs=1000`
25 func TestIntegration(t *testing.T) {
26 eurekaAddr := os.Getenv("EUREKA_ADDR")
27 if eurekaAddr == "" {
28 t.Skip("EUREKA_ADDR is not set")
29 }
30
31 logger := log.NewLogfmtLogger(os.Stderr)
32 logger = log.With(logger, "ts", log.DefaultTimestamp)
33
34 var fargoConfig fargo.Config
35 // Target Eureka server(s).
36 fargoConfig.Eureka.ServiceUrls = []string{eurekaAddr}
37 // How often the subscriber should poll for updates.
38 fargoConfig.Eureka.PollIntervalSeconds = 1
39
40 // Create a Fargo connection and a Eureka registrar.
41 fargoConnection := fargo.NewConnFromConfig(fargoConfig)
42 registrar1 := NewRegistrar(&fargoConnection, instanceTest1, log.With(logger, "component", "registrar1"))
43
44 // Register one instance.
45 registrar1.Register()
46 defer registrar1.Deregister()
47
48 // This should be enough time for the Eureka server response cache to update.
49 time.Sleep(time.Second)
50
51 // Build a Eureka subscriber.
52 factory := func(instance string) (endpoint.Endpoint, io.Closer, error) {
53 t.Logf("factory invoked for %q", instance)
54 return endpoint.Nop, nil, nil
55 }
56 s := NewSubscriber(
57 &fargoConnection,
58 appNameTest,
59 factory,
60 log.With(logger, "component", "subscriber"),
61 )
62 defer s.Stop()
63
64 // We should have one endpoint immediately after subscriber instantiation.
65 endpoints, err := s.Endpoints()
66 if err != nil {
67 t.Error(err)
68 }
69 if want, have := 1, len(endpoints); want != have {
70 t.Errorf("want %d, have %d", want, have)
71 }
72
73 // Register a second instance
74 registrar2 := NewRegistrar(&fargoConnection, instanceTest2, log.With(logger, "component", "registrar2"))
75 registrar2.Register()
76 defer registrar2.Deregister() // In case of exceptional circumstances.
77
78 // This should be enough time for a scheduled update assuming Eureka is
79 // configured with the properties mentioned in the function comments.
80 time.Sleep(2 * time.Second)
81
82 // Now we should have two endpoints.
83 endpoints, err = s.Endpoints()
84 if err != nil {
85 t.Error(err)
86 }
87 if want, have := 2, len(endpoints); want != have {
88 t.Errorf("want %d, have %d", want, have)
89 }
90
91 // Deregister the second instance.
92 registrar2.Deregister()
93
94 // Wait for another scheduled update.
95 time.Sleep(2 * time.Second)
96
97 // And then there was one.
98 endpoints, err = s.Endpoints()
99 if err != nil {
100 t.Error(err)
101 }
102 if want, have := 1, len(endpoints); want != have {
103 t.Errorf("want %d, have %d", want, have)
104 }
105 }
+127
-0
sd/eureka/registrar.go less more
0 package eureka
1
2 import (
3 "fmt"
4 "net/http"
5 "sync"
6 "time"
7
8 "github.com/hudl/fargo"
9
10 "github.com/go-kit/kit/log"
11 "github.com/go-kit/kit/sd"
12 )
13
14 // Matches official Netflix Java client default.
15 const defaultRenewalInterval = 30 * time.Second
16
17 // The methods of fargo.Connection used in this package.
18 type fargoConnection interface {
19 RegisterInstance(instance *fargo.Instance) error
20 DeregisterInstance(instance *fargo.Instance) error
21 ReregisterInstance(instance *fargo.Instance) error
22 HeartBeatInstance(instance *fargo.Instance) error
23 ScheduleAppUpdates(name string, await bool, done <-chan struct{}) <-chan fargo.AppUpdate
24 GetApp(name string) (*fargo.Application, error)
25 }
26
27 type fargoUnsuccessfulHTTPResponse struct {
28 statusCode int
29 messagePrefix string
30 }
31
32 // Registrar maintains service instance liveness information in Eureka.
33 type Registrar struct {
34 conn fargoConnection
35 instance *fargo.Instance
36 logger log.Logger
37 quitc chan chan struct{}
38 sync.Mutex
39 }
40
41 var _ sd.Registrar = (*Registrar)(nil)
42
43 // NewRegistrar returns an Eureka Registrar acting on behalf of the provided
44 // Fargo connection and instance. See the integration test for usage examples.
45 func NewRegistrar(conn fargoConnection, instance *fargo.Instance, logger log.Logger) *Registrar {
46 return &Registrar{
47 conn: conn,
48 instance: instance,
49 logger: log.With(logger, "service", instance.App, "address", fmt.Sprintf("%s:%d", instance.IPAddr, instance.Port)),
50 }
51 }
52
53 // Register implements sd.Registrar.
54 func (r *Registrar) Register() {
55 r.Lock()
56 defer r.Unlock()
57
58 if r.quitc != nil {
59 return // Already in the registration loop.
60 }
61
62 if err := r.conn.RegisterInstance(r.instance); err != nil {
63 r.logger.Log("during", "Register", "err", err)
64 }
65
66 r.quitc = make(chan chan struct{})
67 go r.loop()
68 }
69
70 // Deregister implements sd.Registrar.
71 func (r *Registrar) Deregister() {
72 r.Lock()
73 defer r.Unlock()
74
75 if r.quitc == nil {
76 return // Already deregistered.
77 }
78
79 q := make(chan struct{})
80 r.quitc <- q
81 <-q
82 r.quitc = nil
83 }
84
85 func (r *Registrar) loop() {
86 var renewalInterval time.Duration
87 if r.instance.LeaseInfo.RenewalIntervalInSecs > 0 {
88 renewalInterval = time.Duration(r.instance.LeaseInfo.RenewalIntervalInSecs) * time.Second
89 } else {
90 renewalInterval = defaultRenewalInterval
91 }
92 ticker := time.NewTicker(renewalInterval)
93 defer ticker.Stop()
94
95 for {
96 select {
97 case <-ticker.C:
98 if err := r.heartbeat(); err != nil {
99 r.logger.Log("during", "heartbeat", "err", err)
100 }
101
102 case q := <-r.quitc:
103 if err := r.conn.DeregisterInstance(r.instance); err != nil {
104 r.logger.Log("during", "Deregister", "err", err)
105 }
106 close(q)
107 return
108 }
109 }
110 }
111
112 func (r *Registrar) heartbeat() error {
113 err := r.conn.HeartBeatInstance(r.instance)
114 if err != nil {
115 if u, ok := err.(*fargoUnsuccessfulHTTPResponse); ok && u.statusCode == http.StatusNotFound {
116 // Instance expired (e.g. network partition). Re-register.
117 r.logger.Log("during", "heartbeat", err.Error())
118 return r.conn.ReregisterInstance(r.instance)
119 }
120 }
121 return err
122 }
123
124 func (u *fargoUnsuccessfulHTTPResponse) Error() string {
125 return fmt.Sprintf("err=%s code=%d", u.messagePrefix, u.statusCode)
126 }
+102
-0
sd/eureka/registrar_test.go less more
0 package eureka
1
2 import (
3 "testing"
4 "time"
5 )
6
7 func TestRegistrar(t *testing.T) {
8 connection := &testConnection{
9 errHeartbeat: errTest,
10 }
11
12 registrar1 := NewRegistrar(connection, instanceTest1, loggerTest)
13 registrar2 := NewRegistrar(connection, instanceTest2, loggerTest)
14
15 // Not registered.
16 registrar1.Deregister()
17 if want, have := 0, len(connection.instances); want != have {
18 t.Errorf("want %d, have %d", want, have)
19 }
20
21 // Register.
22 registrar1.Register()
23 if want, have := 1, len(connection.instances); want != have {
24 t.Errorf("want %d, have %d", want, have)
25 }
26
27 registrar2.Register()
28 if want, have := 2, len(connection.instances); want != have {
29 t.Errorf("want %d, have %d", want, have)
30 }
31
32 // Deregister.
33 registrar1.Deregister()
34 if want, have := 1, len(connection.instances); want != have {
35 t.Errorf("want %d, have %d", want, have)
36 }
37
38 // Already registered.
39 registrar1.Register()
40 if want, have := 2, len(connection.instances); want != have {
41 t.Errorf("want %d, have %d", want, have)
42 }
43 registrar1.Register()
44 if want, have := 2, len(connection.instances); want != have {
45 t.Errorf("want %d, have %d", want, have)
46 }
47
48 // Wait for a heartbeat failure.
49 time.Sleep(1010 * time.Millisecond)
50 if want, have := 2, len(connection.instances); want != have {
51 t.Errorf("want %d, have %d", want, have)
52 }
53 registrar1.Deregister()
54 if want, have := 1, len(connection.instances); want != have {
55 t.Errorf("want %d, have %d", want, have)
56 }
57 }
58
59 func TestBadRegister(t *testing.T) {
60 connection := &testConnection{
61 errRegister: errTest,
62 }
63
64 registrar := NewRegistrar(connection, instanceTest1, loggerTest)
65 registrar.Register()
66 if want, have := 0, len(connection.instances); want != have {
67 t.Errorf("want %d, have %d", want, have)
68 }
69 }
70
71 func TestBadDeregister(t *testing.T) {
72 connection := &testConnection{
73 errDeregister: errTest,
74 }
75
76 registrar := NewRegistrar(connection, instanceTest1, loggerTest)
77 registrar.Register()
78 if want, have := 1, len(connection.instances); want != have {
79 t.Errorf("want %d, have %d", want, have)
80 }
81 registrar.Deregister()
82 if want, have := 1, len(connection.instances); want != have {
83 t.Errorf("want %d, have %d", want, have)
84 }
85 }
86
87 func TestExpiredInstance(t *testing.T) {
88 connection := &testConnection{
89 errHeartbeat: errNotFound,
90 }
91
92 registrar := NewRegistrar(connection, instanceTest1, loggerTest)
93 registrar.Register()
94
95 // Wait for a heartbeat failure.
96 time.Sleep(1010 * time.Millisecond)
97
98 if want, have := 1, len(connection.instances); want != have {
99 t.Errorf("want %d, have %d", want, have)
100 }
101 }
+106
-0
sd/eureka/subscriber.go less more
0 package eureka
1
2 import (
3 "fmt"
4
5 "github.com/hudl/fargo"
6
7 "github.com/go-kit/kit/endpoint"
8 "github.com/go-kit/kit/log"
9 "github.com/go-kit/kit/sd"
10 "github.com/go-kit/kit/sd/cache"
11 )
12
13 // Subscriber yields endpoints stored in the Eureka registry for the given app.
14 // Changes in that app are watched and will update the Subscriber endpoints.
15 type Subscriber struct {
16 conn fargoConnection
17 app string
18 factory sd.Factory
19 logger log.Logger
20 cache *cache.Cache
21 quitc chan chan struct{}
22 }
23
24 var _ sd.Subscriber = (*Subscriber)(nil)
25
26 // NewSubscriber returns a Eureka subscriber. It will start watching the given
27 // app string for changes, and update the endpoints accordingly.
28 func NewSubscriber(conn fargoConnection, app string, factory sd.Factory, logger log.Logger) *Subscriber {
29 logger = log.With(logger, "app", app)
30
31 s := &Subscriber{
32 conn: conn,
33 app: app,
34 factory: factory,
35 logger: logger,
36 cache: cache.New(factory, logger),
37 quitc: make(chan chan struct{}),
38 }
39
40 instances, err := s.getInstances()
41 if err == nil {
42 s.logger.Log("instances", len(instances))
43 } else {
44 s.logger.Log("during", "getInstances", "err", err)
45 }
46
47 s.cache.Update(instances)
48 go s.loop()
49 return s
50 }
51
52 // Endpoints implements the Subscriber interface.
53 func (s *Subscriber) Endpoints() ([]endpoint.Endpoint, error) {
54 return s.cache.Endpoints(), nil
55 }
56
57 // Stop terminates the subscriber.
58 func (s *Subscriber) Stop() {
59 q := make(chan struct{})
60 s.quitc <- q
61 <-q
62 s.quitc = nil
63 }
64
65 func (s *Subscriber) loop() {
66 var (
67 await = false
68 done = make(chan struct{})
69 updatec = s.conn.ScheduleAppUpdates(s.app, await, done)
70 )
71 defer close(done)
72
73 for {
74 select {
75 case update := <-updatec:
76 if update.Err != nil {
77 s.logger.Log("during", "Update", "err", update.Err)
78 continue
79 }
80 instances := convertFargoAppToInstances(update.App)
81 s.logger.Log("instances", len(instances))
82 s.cache.Update(instances)
83
84 case q := <-s.quitc:
85 close(q)
86 return
87 }
88 }
89 }
90
91 func (s *Subscriber) getInstances() ([]string, error) {
92 app, err := s.conn.GetApp(s.app)
93 if err != nil {
94 return nil, err
95 }
96 return convertFargoAppToInstances(app), nil
97 }
98
99 func convertFargoAppToInstances(app *fargo.Application) []string {
100 instances := make([]string, len(app.Instances))
101 for i, inst := range app.Instances {
102 instances[i] = fmt.Sprintf("%s:%d", inst.IPAddr, inst.Port)
103 }
104 return instances
105 }
+144
-0
sd/eureka/subscriber_test.go less more
0 package eureka
1
2 import (
3 "io"
4 "testing"
5 "time"
6
7 "github.com/hudl/fargo"
8
9 "github.com/go-kit/kit/endpoint"
10 )
11
12 func TestSubscriber(t *testing.T) {
13 factory := func(string) (endpoint.Endpoint, io.Closer, error) {
14 return endpoint.Nop, nil, nil
15 }
16
17 connection := &testConnection{
18 instances: []*fargo.Instance{instanceTest1},
19 application: appUpdateTest,
20 errApplication: nil,
21 }
22
23 subscriber := NewSubscriber(connection, appNameTest, factory, loggerTest)
24 defer subscriber.Stop()
25
26 endpoints, err := subscriber.Endpoints()
27 if err != nil {
28 t.Fatal(err)
29 }
30
31 if want, have := 1, len(endpoints); want != have {
32 t.Errorf("want %d, have %d", want, have)
33 }
34 }
35
36 func TestSubscriberScheduleUpdates(t *testing.T) {
37 factory := func(string) (endpoint.Endpoint, io.Closer, error) {
38 return endpoint.Nop, nil, nil
39 }
40
41 connection := &testConnection{
42 instances: []*fargo.Instance{instanceTest1},
43 application: appUpdateTest,
44 errApplication: nil,
45 }
46
47 subscriber := NewSubscriber(connection, appNameTest, factory, loggerTest)
48 defer subscriber.Stop()
49
50 endpoints, _ := subscriber.Endpoints()
51 if want, have := 1, len(endpoints); want != have {
52 t.Errorf("want %d, have %d", want, have)
53 }
54
55 time.Sleep(50 * time.Millisecond)
56
57 endpoints, _ = subscriber.Endpoints()
58 if want, have := 2, len(endpoints); want != have {
59 t.Errorf("want %v, have %v", want, have)
60 }
61 }
62
63 func TestBadFactory(t *testing.T) {
64 factory := func(string) (endpoint.Endpoint, io.Closer, error) {
65 return nil, nil, errTest
66 }
67
68 connection := &testConnection{
69 instances: []*fargo.Instance{instanceTest1},
70 application: appUpdateTest,
71 errApplication: nil,
72 }
73
74 subscriber := NewSubscriber(connection, appNameTest, factory, loggerTest)
75 defer subscriber.Stop()
76
77 endpoints, err := subscriber.Endpoints()
78 if err != nil {
79 t.Fatal(err)
80 }
81
82 if want, have := 0, len(endpoints); want != have {
83 t.Errorf("want %d, have %d", want, have)
84 }
85 }
86
87 func TestBadSubscriberInstances(t *testing.T) {
88 factory := func(string) (endpoint.Endpoint, io.Closer, error) {
89 return endpoint.Nop, nil, nil
90 }
91
92 connection := &testConnection{
93 instances: []*fargo.Instance{},
94 errInstances: errTest,
95 application: appUpdateTest,
96 errApplication: nil,
97 }
98
99 subscriber := NewSubscriber(connection, appNameTest, factory, loggerTest)
100 defer subscriber.Stop()
101
102 endpoints, err := subscriber.Endpoints()
103 if err != nil {
104 t.Fatal(err)
105 }
106
107 if want, have := 0, len(endpoints); want != have {
108 t.Errorf("want %d, have %d", want, have)
109 }
110 }
111
112 func TestBadSubscriberScheduleUpdates(t *testing.T) {
113 factory := func(string) (endpoint.Endpoint, io.Closer, error) {
114 return endpoint.Nop, nil, nil
115 }
116
117 connection := &testConnection{
118 instances: []*fargo.Instance{instanceTest1},
119 application: appUpdateTest,
120 errApplication: errTest,
121 }
122
123 subscriber := NewSubscriber(connection, appNameTest, factory, loggerTest)
124 defer subscriber.Stop()
125
126 endpoints, err := subscriber.Endpoints()
127 if err != nil {
128 t.Error(err)
129 }
130 if want, have := 1, len(endpoints); want != have {
131 t.Errorf("want %d, have %d", want, have)
132 }
133
134 time.Sleep(50 * time.Millisecond)
135
136 endpoints, err = subscriber.Endpoints()
137 if err != nil {
138 t.Error(err)
139 }
140 if want, have := 1, len(endpoints); want != have {
141 t.Errorf("want %v, have %v", want, have)
142 }
143 }
+108
-0
sd/eureka/util_test.go less more
0 package eureka
1
2 import (
3 "errors"
4 "reflect"
5
6 "github.com/go-kit/kit/log"
7 "github.com/hudl/fargo"
8 )
9
10 type testConnection struct {
11 instances []*fargo.Instance
12 application *fargo.Application
13 errInstances error
14 errApplication error
15 errHeartbeat error
16 errRegister error
17 errDeregister error
18 }
19
20 var (
21 errTest = errors.New("kaboom")
22 errNotFound = &fargoUnsuccessfulHTTPResponse{statusCode: 404, messagePrefix: "not found"}
23 loggerTest = log.NewNopLogger()
24 appNameTest = "go-kit"
25 appUpdateTest = &fargo.Application{
26 Name: appNameTest,
27 Instances: []*fargo.Instance{instanceTest1, instanceTest2},
28 }
29 instanceTest1 = &fargo.Instance{
30 HostName: "serveregistrar1.acme.org",
31 Port: 8080,
32 App: appNameTest,
33 IPAddr: "192.168.0.1",
34 VipAddress: "192.168.0.1",
35 SecureVipAddress: "192.168.0.1",
36 HealthCheckUrl: "http://serveregistrar1.acme.org:8080/healthz",
37 StatusPageUrl: "http://serveregistrar1.acme.org:8080/status",
38 HomePageUrl: "http://serveregistrar1.acme.org:8080/",
39 Status: fargo.UP,
40 DataCenterInfo: fargo.DataCenterInfo{Name: fargo.MyOwn},
41 LeaseInfo: fargo.LeaseInfo{RenewalIntervalInSecs: 1},
42 }
43 instanceTest2 = &fargo.Instance{
44 HostName: "serveregistrar2.acme.org",
45 Port: 8080,
46 App: appNameTest,
47 IPAddr: "192.168.0.2",
48 VipAddress: "192.168.0.2",
49 SecureVipAddress: "192.168.0.2",
50 HealthCheckUrl: "http://serveregistrar2.acme.org:8080/healthz",
51 StatusPageUrl: "http://serveregistrar2.acme.org:8080/status",
52 HomePageUrl: "http://serveregistrar2.acme.org:8080/",
53 Status: fargo.UP,
54 DataCenterInfo: fargo.DataCenterInfo{Name: fargo.MyOwn},
55 }
56 )
57
58 var _ fargoConnection = (*testConnection)(nil)
59
60 func (c *testConnection) RegisterInstance(i *fargo.Instance) error {
61 if c.errRegister == nil {
62 for _, instance := range c.instances {
63 if reflect.DeepEqual(*instance, *i) {
64 return errors.New("already registered")
65 }
66 }
67
68 c.instances = append(c.instances, i)
69 }
70 return c.errRegister
71 }
72
73 func (c *testConnection) HeartBeatInstance(i *fargo.Instance) error {
74 return c.errHeartbeat
75 }
76
77 func (c *testConnection) DeregisterInstance(i *fargo.Instance) error {
78 if c.errDeregister == nil {
79 var newInstances []*fargo.Instance
80 for _, instance := range c.instances {
81 if reflect.DeepEqual(*instance, *i) {
82 continue
83 }
84 newInstances = append(newInstances, instance)
85 }
86 if len(newInstances) == len(c.instances) {
87 return errors.New("not registered")
88 }
89
90 c.instances = newInstances
91 }
92 return c.errDeregister
93 }
94
95 func (c *testConnection) ReregisterInstance(ins *fargo.Instance) error {
96 return nil
97 }
98
99 func (c *testConnection) ScheduleAppUpdates(name string, await bool, done <-chan struct{}) <-chan fargo.AppUpdate {
100 updatec := make(chan fargo.AppUpdate, 1)
101 updatec <- fargo.AppUpdate{App: c.application, Err: c.errApplication}
102 return updatec
103 }
104
105 func (c *testConnection) GetApp(name string) (*fargo.Application, error) {
106 return &fargo.Application{Name: appNameTest, Instances: c.instances}, c.errInstances
107 }