internal: encapsulate the Contour xDS server

Add a `xds` package that exports a basic interface that xDS servers
can implement. Move the existing Contour xDS server into this package,
and add a factory function to create an instance of it.

After this change, different xDS server implementations can be selected
at startup.

This fixes projectcontour#2775.

Signed-off-by: James Peach <jpeach@vmware.com>

cmd/contour/serve.go

@@ -29,13 +29,13 @@ import (
 	"github.com/projectcontour/contour/internal/contour"
 	"github.com/projectcontour/contour/internal/dag"
 	"github.com/projectcontour/contour/internal/debug"
-	cgrpc "github.com/projectcontour/contour/internal/grpc"
 	"github.com/projectcontour/contour/internal/health"
 	"github.com/projectcontour/contour/internal/httpsvc"
 	"github.com/projectcontour/contour/internal/k8s"
 	"github.com/projectcontour/contour/internal/metrics"
 	"github.com/projectcontour/contour/internal/timeout"
 	"github.com/projectcontour/contour/internal/workgroup"
+	"github.com/projectcontour/contour/internal/xds"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 	"gopkg.in/alecthomas/kingpin.v2"
@@ -408,16 +408,19 @@ func doServe(log logrus.FieldLogger, ctx *serveContext) error {
 	}
 
 	g.Add(func(stop <-chan struct{}) error {
-		log := log.WithField("context", "grpc")
+		log := log.WithField("context", "xds")
 
 		log.Printf("waiting for informer caches to sync")
 		if err := informerSyncList.WaitForSync(stop); err != nil {
 			return err
 		}
 		log.Printf("informer caches synced")
 
-		opts := ctx.grpcOptions()
-		s := cgrpc.NewAPI(log, contour.ResourcesOf(resources), registry, opts...)
+		rpcServer := xds.RegisterServer(
+			xds.NewContourServer(log, contour.ResourcesOf(resources)...),
+			registry,
+			ctx.grpcOptions()...)
+
 		addr := net.JoinHostPort(ctx.xdsAddr, strconv.Itoa(ctx.xdsPort))
 		l, err := net.Listen("tcp", addr)
 		if err != nil {
@@ -434,10 +437,10 @@ func doServe(log logrus.FieldLogger, ctx *serveContext) error {
 
 		go func() {
 			<-stop
-			s.GracefulStop()
+			rpcServer.GracefulStop()
 		}()
 
-		return s.Serve(l)
+		return rpcServer.Serve(l)
 	})
 
 	// Set up SIGTERM handler for graceful shutdown.

go.mod

@@ -19,6 +19,7 @@ require (
 	github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4
 	github.com/prometheus/common v0.6.0
 	github.com/sirupsen/logrus v1.4.2
+	github.com/stretchr/testify v1.5.1 // indirect
 	golang.org/x/tools v0.0.0-20190929041059-e7abfedfabcf // indirect
 	google.golang.org/grpc v1.25.1
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6

go.sum

@@ -357,6 +357,8 @@ github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=

internal/contour/resources.go

@@ -18,20 +18,20 @@ package contour
 
 import (
 	"github.com/projectcontour/contour/internal/dag"
-	"github.com/projectcontour/contour/internal/grpc"
+	"github.com/projectcontour/contour/internal/xds"
 )
 
 // ResourceCache is a store of an xDS resource type. It is able to
 // visit the dag.DAG to update the its resource collection, then
 // serve those resources over xDS.
 type ResourceCache interface {
 	dag.Observer
-	grpc.Resource
+	xds.Resource
 }
 
-// ResourcesOf transliterates a slice of ResourceCache into a slice of grpc.Resource.
-func ResourcesOf(in []ResourceCache) []grpc.Resource {
-	out := make([]grpc.Resource, len(in))
+// ResourcesOf transliterates a slice of ResourceCache into a slice of xds.Resource.
+func ResourcesOf(in []ResourceCache) []xds.Resource {
+	out := make([]xds.Resource, len(in))
 	for i := range in {
 		out[i] = in[i]
 	}

internal/contour/server_test.go

@@ -24,8 +24,7 @@ import (
 	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v2"
 	resource "github.com/envoyproxy/go-control-plane/pkg/resource/v2"
 	"github.com/projectcontour/contour/internal/dag"
-	cgrpc "github.com/projectcontour/contour/internal/grpc"
-	"github.com/prometheus/client_golang/prometheus"
+	"github.com/projectcontour/contour/internal/xds"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -197,15 +196,15 @@ func TestGRPC(t *testing.T) {
 				&SecretCache{},
 				&RouteCache{},
 				&ClusterCache{},
+				et,
 			}
 
 			eh = &EventHandler{
 				Observer:    dag.ComposeObservers(ObserversOf(resources)...),
 				FieldLogger: log,
 			}
 
-			r := prometheus.NewRegistry()
-			srv := cgrpc.NewAPI(log, append(ResourcesOf(resources), et), r)
+			srv := xds.RegisterServer(xds.NewContourServer(log, ResourcesOf(resources)...), nil)
 			l, err := net.Listen("tcp", "127.0.0.1:0")
 			check(t, err)
 			done := make(chan error, 1)

internal/featuretests/featuretests.go

@@ -32,12 +32,12 @@ import (
 	"github.com/projectcontour/contour/internal/contour"
 	"github.com/projectcontour/contour/internal/dag"
 	"github.com/projectcontour/contour/internal/fixture"
-	cgrpc "github.com/projectcontour/contour/internal/grpc"
 	"github.com/projectcontour/contour/internal/k8s"
 	"github.com/projectcontour/contour/internal/metrics"
 	"github.com/projectcontour/contour/internal/protobuf"
 	"github.com/projectcontour/contour/internal/sorter"
 	"github.com/projectcontour/contour/internal/workgroup"
+	"github.com/projectcontour/contour/internal/xds"
 	"github.com/prometheus/client_golang/prometheus"
 	"google.golang.org/grpc"
 	v1 "k8s.io/api/core/v1"
@@ -75,6 +75,7 @@ func setup(t *testing.T, opts ...interface{}) (cache.ResourceEventHandler, *Cont
 		&contour.SecretCache{},
 		&contour.RouteCache{},
 		&contour.ClusterCache{},
+		et,
 	}
 
 	r := prometheus.NewRegistry()
@@ -111,8 +112,10 @@ func setup(t *testing.T, opts ...interface{}) (cache.ResourceEventHandler, *Cont
 
 	l, err := net.Listen("tcp", "127.0.0.1:0")
 	check(t, err)
-	// Resource types in xDS v2.
-	srv := cgrpc.NewAPI(log, append(contour.ResourcesOf(resources), et), r)
+
+	srv := xds.RegisterServer(
+		xds.NewContourServer(log, contour.ResourcesOf(resources)...),
+		r /* Prometheus registry */)
 
 	var g workgroup.Group