enable envoy configs to be grouped

Gopkg.toml

@@ -82,7 +82,7 @@
   name = "github.com/golang/glog"
 
 [[constraint]]
-  branch = "master"
+  version = "v1.1.0"
   name = "github.com/golang/protobuf"
 
 [[constraint]]

api/v1/virtualmesh.proto

@@ -14,7 +14,7 @@ import "metadata.proto";
  * A Virtual Mesh is a container for a set of Virtual Services that will be used to generate a single proxy config
  * to be applied to one or more Envoy nodes. The Virtual Mesh is best understood as an in-mesh application's localized view
  * of the rest of the mesh.
- * Each domains for each Virtual Services contained in a Virtual Mesh cannot appear more than once, or the Virtual Mesh
+ * Each domain for each Virtual Service contained in a Virtual Mesh cannot appear more than once, or the Virtual Mesh
  * will be invalid.
  */
 message VirtualMesh {
@@ -28,7 +28,7 @@ message VirtualMesh {
     // One or more lowercase rfc1035/rfc1123 labels separated by '.' with a maximum length of 253 characters.
     string name = 1;
 
-    // the list of names of the virtual services this vmesh includes.
+    // the list of names of the virtual services this vmesh encapsulates.
     repeated string virtual_services = 2;
 
     // Status indicates the validation status of the virtual mesh resource.

docs/api.json

@@ -370,7 +370,7 @@
           "name": "VirtualMesh",
           "longName": "VirtualMesh",
           "fullName": "v1.VirtualMesh",
-          "description": "A Virtual Mesh is a container for a set of Virtual Services that will be used to generate a single proxy config\nto be applied to one or more Envoy nodes. The Virtual Mesh is best understood as an in-mesh application's localized view\nof the rest of the mesh.\nEach domains for each Virtual Services contained in a Virtual Mesh cannot appear more than once, or the Virtual Mesh\nwill be invalid.",
+          "description": "A Virtual Mesh is a container for a set of Virtual Services that will be used to generate a single proxy config\nto be applied to one or more Envoy nodes. The Virtual Mesh is best understood as an in-mesh application's localized view\nof the rest of the mesh.\nEach domain for each Virtual Service contained in a Virtual Mesh cannot appear more than once, or the Virtual Mesh\nwill be invalid.",
           "hasExtensions": false,
           "hasFields": true,
           "extensions": [],
@@ -386,7 +386,7 @@
             },
             {
               "name": "virtual_services",
-              "description": "the list of names of the virtual services this vmesh includes.",
+              "description": "the list of names of the virtual services this vmesh encapsulates.",
               "label": "repeated",
               "type": "string",
               "longType": "string",

example/nats/envoy.yaml

@@ -1,7 +1,7 @@
 
 node:
   cluster: ingress
-  id: ingress-1
+  id: ingress~1
 static_resources:
   clusters:
   - name: xds_cluster

hack/run-local-binaries.sh

@@ -38,7 +38,7 @@ GLOO_IP=localhost
 cat > ${CONFIG_DIR}/envoy.yaml <<EOF
 node:
   cluster: ingress
-  id: ingress
+  id: ingress~1
 
 static_resources:
   clusters:

hack/run-local-docker.sh

@@ -50,7 +50,7 @@ CONTROL_PLANE_IP=localhost
 cat > ${CONFIG_DIR}/envoy.yaml <<EOF
 node:
   cluster: ingress
-  id: ingress
+  id: ingress~1
 
 static_resources:
   clusters:

hack/run-local.sh

@@ -38,7 +38,7 @@ GLOO_IP=127.0.0.1
 cat > ${CONFIG_DIR}/envoy.yaml <<EOF
 node:
   cluster: ingress
-  id: ingress
+  id: ingress~1
 
 static_resources:
   clusters:

install/helm/gloo/templates/ingress-configmap.yaml

@@ -7,7 +7,7 @@ data:
   envoy.yaml: |
     node:
       cluster: ingress
-      id: NODE_ID_PLACE_HOLDER
+      id: ingress~NODE_ID_PLACE_HOLDER
     static_resources:
       clusters:
       - name: xds_cluster

install/kube/install.yaml

@@ -46,7 +46,7 @@ data:
   envoy.yaml: |
     node:
       cluster: ingress
-      id: NODE_ID_PLACE_HOLDER
+      id: ingress~NODE_ID_PLACE_HOLDER
     static_resources:
       clusters:
       - name: xds_cluster

install/nomad/install.nomad

@@ -105,7 +105,7 @@ job "gloo" {
         data = <<EOF
 node:
   cluster: ingress
-  id: {{ env "NOMAD_ALLOC_ID" }}
+  id: ingress~{{ env "NOMAD_ALLOC_ID" }}
 
 static_resources:
   clusters:

install/openshift/install.yml

@@ -43,7 +43,7 @@ data:
   envoy.yaml: |
     node:
       cluster: ingress
-      id: NODE_ID_PLACE_HOLDER
+      id: ingress~NODE_ID_PLACE_HOLDER
     static_resources:
       clusters:
       - name: xds_cluster

internal/control-plane/bootstrap/flags/ingress_flags.go

@@ -7,7 +7,7 @@ import (
 
 func AddIngressFlags(cmd *cobra.Command, opts *bootstrap.Options) {
 	// TODO ingress.bind-adress
-	cmd.PersistentFlags().StringVar(&opts.IngressOptions.BindAddress, "envoy.bind-adress", "", "The address that the ingress envoy should bind to.")
-	cmd.PersistentFlags().IntVar(&opts.IngressOptions.Port, "envoy.port", 8080, "The HTTP port envoy uses.")
-	cmd.PersistentFlags().IntVar(&opts.IngressOptions.SecurePort, "envoy.secure-port", 8443, "The HTTPS port envoy uses.")
+	cmd.PersistentFlags().StringVar(&opts.IngressOptions.BindAddress, "envoy.bind-adress", "::", "The address that the ingress envoy should bind to.")
+	cmd.PersistentFlags().Uint32Var(&opts.IngressOptions.Port, "envoy.port", 8080, "The HTTP port envoy uses.")
+	cmd.PersistentFlags().Uint32Var(&opts.IngressOptions.SecurePort, "envoy.secure-port", 8443, "The HTTPS port envoy uses.")
 }

internal/control-plane/bootstrap/options.go

@@ -11,6 +11,6 @@ type Options struct {
 
 type IngressOptions struct {
 	BindAddress string
-	Port        int
-	SecurePort  int
+	Port        uint32
+	SecurePort  uint32
 }

internal/control-plane/configwatcher/config_watcher.go

@@ -34,11 +34,17 @@ func NewConfigWatcher(storageClient storage.Interface) (*configWatcher, error) {
 		log.Warnf("Startup: failed to read virtual services from storage: %v", err)
 		initialVirtualServices = []*v1.VirtualService{}
 	}
+	initialVirtualMeshes, err := storageClient.V1().VirtualMeshes().List()
+	if err != nil {
+		log.Warnf("Startup: failed to read virtual services from storage: %v", err)
+		initialVirtualMeshes = []*v1.VirtualMesh{}
+	}
 	configs := make(chan *v1.Config)
 	// do a first time read
 	cache := &v1.Config{
 		Upstreams:       initialUpstreams,
 		VirtualServices: initialVirtualServices,
+		VirtualMeshes: initialVirtualMeshes,
 	}
 	// throw it down the channel to get things going
 	go func() {
@@ -67,6 +73,7 @@ func NewConfigWatcher(storageClient storage.Interface) (*configWatcher, error) {
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create watcher for upstreams")
 	}
+
 	syncvServices := func(updatedList []*v1.VirtualService, _ *v1.VirtualService) {
 		sort.SliceStable(updatedList, func(i, j int) bool {
 			return updatedList[i].GetName() < updatedList[j].GetName()
@@ -90,8 +97,32 @@ func NewConfigWatcher(storageClient storage.Interface) (*configWatcher, error) {
 		return nil, errors.Wrap(err, "failed to create watcher for virtualservices")
 	}
 
+
+	syncvMeshes := func(updatedList []*v1.VirtualMesh, _ *v1.VirtualMesh) {
+		sort.SliceStable(updatedList, func(i, j int) bool {
+			return updatedList[i].GetName() < updatedList[j].GetName()
+		})
+
+		diff, equal := messagediff.PrettyDiff(cache.VirtualMeshes, updatedList)
+		if equal {
+			return
+		}
+		log.GreyPrintf("change detected in virtualservices: %v", diff)
+
+		cache.VirtualMeshes = updatedList
+		configs <- cache
+	}
+	vMeshWatcher, err := storageClient.V1().VirtualMeshes().Watch(&storage.VirtualMeshEventHandlerFuncs{
+		AddFunc:    syncvMeshes,
+		UpdateFunc: syncvMeshes,
+		DeleteFunc: syncvMeshes,
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create watcher for virtualservices")
+	}
+
 	return &configWatcher{
-		watchers: []*storage.Watcher{vServiceWatcher, upstreamWatcher},
+		watchers: []*storage.Watcher{vServiceWatcher, vMeshWatcher, upstreamWatcher},
 		configs:  configs,
 		errs:     make(chan error),
 	}, nil

internal/control-plane/configwatcher/crd_config_watcher_test.go

@@ -35,7 +35,36 @@ var _ = Describe("KubeConfigWatcher", func() {
 		TeardownKube(namespace)
 	})
 	Describe("controller", func() {
-		It("watches kube crds", func() {
+		It("watches kube upstream crds", func() {
+			cfg, err := clientcmd.BuildConfigFromFlags(masterUrl, kubeconfigPath)
+			Expect(err).NotTo(HaveOccurred())
+
+			storageClient, err := crd.NewStorage(cfg, namespace, time.Second)
+			Expect(err).NotTo(HaveOccurred())
+
+			watcher, err := NewConfigWatcher(storageClient)
+			Must(err)
+			go func() { watcher.Run(make(chan struct{})) }()
+
+			upstream := NewTestUpstream1()
+			created, err := storageClient.V1().Upstreams().Create(upstream)
+			Expect(err).NotTo(HaveOccurred())
+
+			// give controller time to register
+			time.Sleep(time.Second * 2)
+
+			select {
+			case <-time.After(time.Second * 5):
+				Expect(fmt.Errorf("expected to have received resource event before 5s")).NotTo(HaveOccurred())
+			case cfg := <-watcher.Config():
+				Expect(len(cfg.Upstreams)).To(Equal(1))
+				Expect(cfg.Upstreams[0]).To(Equal(created))
+				Expect(cfg.Upstreams[0].Spec).To(Equal(created.Spec))
+			case err := <-watcher.Error():
+				Expect(err).NotTo(HaveOccurred())
+			}
+		})
+		It("watches kube virtual service crds", func() {
 			cfg, err := clientcmd.BuildConfigFromFlags(masterUrl, kubeconfigPath)
 			Expect(err).NotTo(HaveOccurred())
 
@@ -65,5 +94,35 @@ var _ = Describe("KubeConfigWatcher", func() {
 				Expect(err).NotTo(HaveOccurred())
 			}
 		})
+		It("watches kube virtual mesh crds", func() {
+			cfg, err := clientcmd.BuildConfigFromFlags(masterUrl, kubeconfigPath)
+			Expect(err).NotTo(HaveOccurred())
+
+			storageClient, err := crd.NewStorage(cfg, namespace, time.Second)
+			Expect(err).NotTo(HaveOccurred())
+
+			watcher, err := NewConfigWatcher(storageClient)
+			Must(err)
+			go func() { watcher.Run(make(chan struct{})) }()
+
+			virtualMesh := NewTestVirtualMesh("something", "foo")
+			created, err := storageClient.V1().VirtualMeshes().Create(virtualMesh)
+			Expect(err).NotTo(HaveOccurred())
+
+			// give controller time to register
+			time.Sleep(time.Second * 2)
+
+			select {
+			case <-time.After(time.Second * 5):
+				Expect(fmt.Errorf("expected to have received resource event before 5s")).NotTo(HaveOccurred())
+			case cfg := <-watcher.Config():
+				Expect(len(cfg.VirtualMeshes)).To(Equal(1))
+				Expect(cfg.VirtualMeshes[0]).To(Equal(created))
+				Expect(len(cfg.VirtualMeshes[0].VirtualServices)).To(Equal(1))
+				Expect(cfg.VirtualMeshes[0].VirtualServices[0]).To(Equal(created.VirtualServices[0]))
+			case err := <-watcher.Error():
+				Expect(err).NotTo(HaveOccurred())
+			}
+		})
 	})
 })

internal/control-plane/eventloop/eventloop.go

@@ -34,14 +34,6 @@ type eventLoop struct {
 	startFuncs []func() error
 }
 
-func translatorConfig(opts bootstrap.Options) translator.TranslatorConfig {
-	return translator.TranslatorConfig{
-		IngressBindAddress: opts.IngressOptions.BindAddress,
-		IngressPort:        uint32(opts.IngressOptions.Port),
-		IngressSecurePort:  uint32(opts.IngressOptions.SecurePort),
-	}
-}
-
 func Setup(opts bootstrap.Options, xdsPort int, stop <-chan struct{}) (*eventLoop, error) {
 	store, err := configstorage.Bootstrap(opts.Options)
 	if err != nil {
@@ -63,14 +55,17 @@ func Setup(opts bootstrap.Options, xdsPort int, stop <-chan struct{}) (*eventLoo
 		return nil, errors.Wrap(err, "failed to set up file watcher")
 	}
 
-	xdsConfig, _, err := xds.RunXDS(xdsPort)
+	// create a snapshot to give to misconfigured envoy instances
+	badNodeSnapshot := xds.BadNodeSnapshot(opts.IngressOptions.BindAddress, opts.IngressOptions.Port)
+
+	xdsConfig, _, err := xds.RunXDS(xdsPort, badNodeSnapshot)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to start xds server")
 	}
 
 	plugs := plugins.RegisteredPlugins()
 
-	trans := translator.NewTranslator(translatorConfig(opts), plugs)
+	trans := translator.NewTranslator(opts.IngressOptions, plugs)
 
 	e := &eventLoop{
 		configWatcher:   cfgWatcher,
@@ -227,8 +222,8 @@ func (e *eventLoop) updateXds(cache *cache) {
 		}
 	}
 
-	log.Debugf("FINAL: XDS Snapshot: %v", snapshot)
-	e.xdsConfig.SetSnapshot(xds.NodeKey, *snapshot)
+	log.Debugf("Setting xDS Snapshot for Virtual Mesh %v: %v", "ingress", snapshot)
+	e.xdsConfig.SetSnapshot("ingress", *snapshot)
 }
 
 // fan out to cover all endpoint discovery services
@@ -268,7 +263,7 @@ func (e *eventLoop) errors() <-chan error {
 	for _, ed := range e.endpointDiscoveries {
 		go func() {
 			for err := range ed.Error() {
-				aggregatedErrorsChan <- err
+				aggregatedErrorsChan <-  errors.Wrap(err, "endpoint discovery encountered an error")
 			}
 		}()
 	}