Use Envoy Aggregated Discovery Service via the gRPC API

.travis.yml

@@ -1,9 +1,8 @@
-language: go
+os: linux
 
+language: go
 go:
-  - 1.12.x
-
-sudo: required
+  - 1.13.x
 
 services:
   - docker
@@ -12,7 +11,7 @@ env:
   - GO111MODULE=on
 
 before_install:
-  - sudo apt-get install -y nodejs
+  - nvm install node
   - curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | sh -s -- -b $(go env GOPATH)/bin v1.23.1
 
 script:

README.md

@@ -21,9 +21,9 @@ or can also leverage:
 
  * [Lyft's Envoy Proxy](https://github.com/envoyproxy/envoy) - In less than
    a year it is fast becoming a core microservices architecture component.
-   Sidecar implements the Envoy proxy SDS, CDS, and LDS APIs (v1). These
-   allow a standalone Envoy to be entirely configured by Sidecar. This is
-   best used with Nitro's
+   Sidecar implements the Envoy proxy SDS, CDS, LDS (V1) and gRPC (V2) APIs.
+   These allow a standalone Envoy to be entirely configured by Sidecar. This
+   is best used with Nitro's
    [Envoy proxy container](https://hub.docker.com/r/gonitro/envoyproxy/tags/).
 
  * [haproxy-api](https://github.com/Nitro/haproxy-api) - A separation layer
@@ -229,6 +229,12 @@ Defaults are in bold at the end of the line:
  * `HAPROXY_USE_HOSTNAMES`: Should we write hostnames in the HAproxy config instead
    of IP addresses? **`false`**
 
+ * `ENVOY_USE_GRPC_API`: Enable the Envoy gRPC API (V2) **`true`**
+ * `ENVOY_BIND_IP`: The IP that Envoy should bind to on the host **192.168.168.168**
+ * `ENVOY_USE_HOSTNAMES`: Should we write hostnames in the Envoy config instead
+   of IP addresses? **`false`**
+ * `ENVOY_GRPC_PORT`: The port for the Envoy API gRPC server **`7776`**
+
 
 ### Ports
 
@@ -424,7 +430,7 @@ The logging output is pretty good in the normal `info` level. It can be made
 quite verbose in `debug` mode, and contains lots of information about what's
 going on and what the current state is. The web interface also contains a lot
 of runtime information on the cluster and the services. If you are running
-HAproxy, it's also recommneded that you expose the HAproxy stats port on 3212
+HAproxy, it's also recommended that you expose the HAproxy stats port on 3212
 so that Sidecar can find it.
 
 Currently the web interface runs on port 7777 on each machine that runs
@@ -457,11 +463,18 @@ Envoy Proxy Support
 -------------------
 
 Envoy uses a very different model than HAproxy and thus Sidecar's support for
-it is quite different from its support for HAproxy. Envoy makes requests to a
-variety of discovery service APIs on a timed basis. Sidecar currently
-implements three of these: the Cluster Discovery Service (CDS), the Service
-Discovery Service (SDS), and the Listeners Discovery Service (LDS). Nitro
-builds and supports [an Envoy
+it is quite different from its support for HAproxy.
+
+When using the REST-based LDS API (V1), Envoy makes requests to a variety of
+discovery service APIs on a timed basis. Sidecar currently implements three
+of these: the Cluster Discovery Service (CDS), the Service Discovery Service
+(SDS), and the Listeners Discovery Service (LDS). When using the gRPC V2 API,
+Sidecar sends updates to Envoy as soon as possible via gRPC.
+
+Note that the LDS API (V1) has been deprecated by Envoy and it's recommended
+to use the gRPC-based V2 API.
+
+Nitro builds and supports [an Envoy
 container](https://hub.docker.com/r/gonitro/envoyproxy/tags/) that is tested
 and works against Sidecar. This is the easiest way to run Envoy with Sidecar.
 You can find an example container configuration

catalog/services_state.go

@@ -24,15 +24,16 @@ import (
 // servers to Service lists and manages the lifecycle.
 
 const (
-	TOMBSTONE_LIFESPAN       = 3 * time.Hour                  // How long we keep tombstones around
-	TOMBSTONE_COUNT          = 10                             // Send tombstones at 1 per second 10 times
-	ALIVE_COUNT              = 5                              // Send new services at 1 per second 5 times
-	TOMBSTONE_SLEEP_INTERVAL = 2 * time.Second                // Sleep between local service checks
-	TOMBSTONE_RETRANSMIT     = 1 * time.Second                // Time between tombstone retranmission
-	ALIVE_LIFESPAN           = 1*time.Minute + 20*time.Second // Down if not heard from in 80 seconds
-	DRAINING_LIFESPAN        = 10 * time.Minute               // Down if not heard from in 10 minutes
-	ALIVE_SLEEP_INTERVAL     = 1 * time.Second                // Sleep between local service checks
-	ALIVE_BROADCAST_INTERVAL = 1 * time.Minute                // Broadcast Alive messages every minute
+	TOMBSTONE_LIFESPAN         = 3 * time.Hour                  // How long we keep tombstones around
+	TOMBSTONE_COUNT            = 10                             // Send tombstones at 1 per second 10 times
+	ALIVE_COUNT                = 5                              // Send new services at 1 per second 5 times
+	TOMBSTONE_SLEEP_INTERVAL   = 2 * time.Second                // Sleep between local service checks
+	TOMBSTONE_RETRANSMIT       = 1 * time.Second                // Time between tombstone retranmission
+	ALIVE_LIFESPAN             = 1*time.Minute + 20*time.Second // Down if not heard from in 80 seconds
+	DRAINING_LIFESPAN          = 10 * time.Minute               // Down if not heard from in 10 minutes
+	ALIVE_SLEEP_INTERVAL       = 1 * time.Second                // Sleep between local service checks
+	ALIVE_BROADCAST_INTERVAL   = 1 * time.Minute                // Broadcast Alive messages every minute
+	LISTENER_EVENT_BUFFER_SIZE = 20                             // The number of events that can be buffered in the listener eventChannel
 )
 
 // A ChangeEvent represents the time and hostname that was modified and signals a major
@@ -147,6 +148,9 @@ func (state *ServicesState) HasServer(hostname string) bool {
 
 // A server has left the cluster, so tombstone all of its records
 func (state *ServicesState) ExpireServer(hostname string) {
+	state.Lock()
+	defer state.Unlock()
+
 	if !state.HasServer(hostname) || len(state.Servers[hostname].Services) == 0 {
 		log.Infof("No records to expire for %s", hostname)
 		return
@@ -171,8 +175,8 @@ func (state *ServicesState) ExpireServer(hostname string) {
 
 	for _, svc := range state.Servers[hostname].Services {
 		previousStatus := svc.Status
-		state.ServiceChanged(svc, previousStatus, svc.Updated)
 		svc.Tombstone()
+		state.ServiceChanged(svc, previousStatus, svc.Updated)
 		tombstones = append(tombstones, *svc)
 	}
 

catalog/services_state_test.go

@@ -670,13 +670,25 @@ func Test_ClusterMembershipManagement(t *testing.T) {
 				state.AddServiceEntry(service1)
 				state.AddServiceEntry(service2)
 
+				dummyListener := mockListener{
+					events: make(chan ChangeEvent, len(state.Servers[hostname].Services)),
+				}
+				state.AddListener(&dummyListener)
+
 				go state.ExpireServer(hostname)
 				expired := <-state.Broadcasts
 
 				So(len(expired), ShouldEqual, 2)
 				// Timestamps chagne when tombstoning, so regex match
 				So(expired[0], ShouldMatch, "^{\"ID\":\"deadbeef.*\"Status\":1}$")
 				So(expired[1], ShouldMatch, "^{\"ID\":\"deadbeef.*\"Status\":1}$")
+
+				Convey("and sends the tombstones to any listener", func() {
+					for i := 0; i < len(state.Servers[hostname].Services); i++ {
+						changeEvent := <-dummyListener.Chan()
+						So(changeEvent.Service.Status, ShouldEqual, service.TOMBSTONE)
+					}
+				})
 			})
 
 			Convey("does not announce services for hosts with none", func() {

catalog/url_listener.go

@@ -71,7 +71,7 @@ func NewUrlListener(listenurl string, managed bool) *UrlListener {
 		Url:          listenurl,
 		looper:       director.NewFreeLooper(director.FOREVER, errorChan),
 		Client:       &http.Client{Timeout: ClientTimeout, Jar: cookieJar},
-		eventChannel: make(chan ChangeEvent, 20),
+		eventChannel: make(chan ChangeEvent, LISTENER_EVENT_BUFFER_SIZE),
 		Retries:      DefaultRetries,
 		managed:      managed,
 		name:         "UrlListener(" + listenurl + ")",

config.go → config/config.go

@@ -1,9 +1,10 @@
-package main
+package config
 
 import (
 	"time"
 
 	"github.com/kelseyhightower/envconfig"
+	log "github.com/sirupsen/logrus"
 	"gopkg.in/relistan/rubberneck.v1"
 )
 
@@ -24,6 +25,13 @@ type HAproxyConfig struct {
 	UseHostnames bool   `envconfig:"USE_HOSTNAMES"`
 }
 
+type EnvoyConfig struct {
+	UseGRPCAPI   bool   `envconfig:"USE_GRPC_API" default:"true"`
+	BindIP       string `envconfig:"BIND_IP" default:"192.168.168.168"`
+	UseHostnames bool   `envconfig:"USE_HOSTNAMES"`
+	GRPCPort     string `envconfig:"GRPC_PORT" default:"7776"`
+}
+
 type ServicesConfig struct {
 	NameMatch    string `envconfig:"NAME_MATCH"`
 	ServiceNamer string `envconfig:"NAMER" default:"docker_label"`
@@ -59,10 +67,11 @@ type Config struct {
 	StaticDiscovery StaticConfig       // STATIC_
 	Services        ServicesConfig     // SERVICES_
 	HAproxy         HAproxyConfig      // HAPROXY_
+	Envoy           EnvoyConfig        // ENVOY_
 	Listeners       ListenerUrlsConfig // LISTENERS_
 }
 
-func parseConfig() Config {
+func ParseConfig() *Config {
 	var config Config
 
 	errs := []error{
@@ -71,15 +80,16 @@ func parseConfig() Config {
 		envconfig.Process("static", &config.StaticDiscovery),
 		envconfig.Process("services", &config.Services),
 		envconfig.Process("haproxy", &config.HAproxy),
+		envconfig.Process("envoy", &config.Envoy),
 		envconfig.Process("listeners", &config.Listeners),
 	}
 
 	for _, err := range errs {
 		if err != nil {
 			rubberneck.Print(config)
-			exitWithError(err, "Can't parse environment config!")
+			log.Fatalf("Can't parse environment config: %s", err)
 		}
 	}
 
-	return config
+	return &config
 }