github
/
consul
mirror of https://github.com/hashicorp/consul
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #7533 from hashicorp/dnephin/xds-server-1 

agent/xds: small cleanup
pull/7570/head
Daniel Nephin 2020-04-01 11:24:50 -04:00 committed by GitHub
parent efaf0f9a18 1d90ecc31d
commit 190fc3c732
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 24 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
agent/xds/clusters.go
View File

@ -22,16 +22,16 @@ import (
)
// clustersFromSnapshot returns the xDS API representation of the "clusters" in the snapshot.
func (s *Server) clustersFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func (s *Server) clustersFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, _ string) ([]proto.Message, error) {
if cfgSnap == nil {
return nil, errors.New("nil config given")
}
switch cfgSnap.Kind {
case structs.ServiceKindConnectProxy:
return s.clustersFromSnapshotConnectProxy(cfgSnap, token)
return s.clustersFromSnapshotConnectProxy(cfgSnap)
case structs.ServiceKindMeshGateway:
return s.clustersFromSnapshotMeshGateway(cfgSnap, token)
return s.clustersFromSnapshotMeshGateway(cfgSnap)
default:
return nil, fmt.Errorf("Invalid service kind: %v", cfgSnap.Kind)
}
@ -39,7 +39,7 @@ func (s *Server) clustersFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token st
// clustersFromSnapshot returns the xDS API representation of the "clusters"
// (upstreams) in the snapshot.
func (s *Server) clustersFromSnapshotConnectProxy(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func (s *Server) clustersFromSnapshotConnectProxy(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
// TODO(rb): this sizing is a low bound.
clusters := make([]proto.Message, 0, len(cfgSnap.Proxy.Upstreams)+1)
@ -112,7 +112,7 @@ func makeExposeClusterName(destinationPort int) string {
// clustersFromSnapshotMeshGateway returns the xDS API representation of the "clusters"
// for a mesh gateway. This will include 1 cluster per remote datacenter as well as
// 1 cluster for each service subset.
func (s *Server) clustersFromSnapshotMeshGateway(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func (s *Server) clustersFromSnapshotMeshGateway(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
datacenters := cfgSnap.MeshGateway.Datacenters()
// 1 cluster per remote dc + 1 cluster per local service (this is a lower bound - all subset specific clusters will be appended)

29
agent/xds/clusters_test.go
View File

@ -505,35 +505,6 @@ type customClusterJSONOptions struct {
TLSContext string
}
var customEDSClusterJSONTpl = `{
{{ if .IncludeType -}}
"@type": "type.googleapis.com/envoy.api.v2.Cluster",
{{- end }}
{{ if .TLSContext -}}
"tlsContext": {{ .TLSContext }},
{{- end }}
"name": "{{ .Name }}",
"type": "EDS",
"edsClusterConfig": {
"edsConfig": {
"ads": {
}
}
},
"connectTimeout": "5s"
}`
var customEDSClusterJSONTemplate = template.Must(template.New("").Parse(customEDSClusterJSONTpl))
func customEDSClusterJSON(t *testing.T, opts customClusterJSONOptions) string {
t.Helper()
var buf bytes.Buffer
err := customEDSClusterJSONTemplate.Execute(&buf, opts)
require.NoError(t, err)
return buf.String()
}
var customAppClusterJSONTpl = `{
{{ if .IncludeType -}}
"@type": "type.googleapis.com/envoy.api.v2.Cluster",

10
agent/xds/endpoints.go
View File

@ -22,16 +22,16 @@ const (
)
// endpointsFromSnapshot returns the xDS API representation of the "endpoints"
func (s *Server) endpointsFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func (s *Server) endpointsFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, _ string) ([]proto.Message, error) {
if cfgSnap == nil {
return nil, errors.New("nil config given")
}
switch cfgSnap.Kind {
case structs.ServiceKindConnectProxy:
return s.endpointsFromSnapshotConnectProxy(cfgSnap, token)
return s.endpointsFromSnapshotConnectProxy(cfgSnap)
case structs.ServiceKindMeshGateway:
return s.endpointsFromSnapshotMeshGateway(cfgSnap, token)
return s.endpointsFromSnapshotMeshGateway(cfgSnap)
default:
return nil, fmt.Errorf("Invalid service kind: %v", cfgSnap.Kind)
}
@ -39,7 +39,7 @@ func (s *Server) endpointsFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token s
// endpointsFromSnapshotConnectProxy returns the xDS API representation of the "endpoints"
// (upstream instances) in the snapshot.
func (s *Server) endpointsFromSnapshotConnectProxy(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func (s *Server) endpointsFromSnapshotConnectProxy(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
resources := make([]proto.Message, 0,
len(cfgSnap.ConnectProxy.PreparedQueryEndpoints)+len(cfgSnap.ConnectProxy.WatchedUpstreamEndpoints))
@ -170,7 +170,7 @@ func (s *Server) filterSubsetEndpoints(subset *structs.ServiceResolverSubset, en
return endpoints, nil
}
func (s *Server) endpointsFromSnapshotMeshGateway(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func (s *Server) endpointsFromSnapshotMeshGateway(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
datacenters := cfgSnap.MeshGateway.Datacenters()
resources := make([]proto.Message, 0, len(datacenters)+len(cfgSnap.MeshGateway.ServiceGroups))

4
agent/xds/listeners.go
View File

@ -39,7 +39,7 @@ func (s *Server) listenersFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token s
case structs.ServiceKindConnectProxy:
return s.listenersFromSnapshotConnectProxy(cfgSnap, token)
case structs.ServiceKindMeshGateway:
return s.listenersFromSnapshotMeshGateway(cfgSnap, token)
return s.listenersFromSnapshotMeshGateway(cfgSnap)
default:
return nil, fmt.Errorf("Invalid service kind: %v", cfgSnap.Kind)
}
@ -180,7 +180,7 @@ func parseCheckPath(check structs.CheckType) (structs.ExposePath, error) {
}
// listenersFromSnapshotMeshGateway returns the "listener" for a mesh-gateway service
func (s *Server) listenersFromSnapshotMeshGateway(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func (s *Server) listenersFromSnapshotMeshGateway(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
cfg, err := ParseGatewayConfig(cfgSnap.Proxy.Config)
if err != nil {
// Don't hard fail on a config typo, just warn. The parse func returns

6
agent/xds/routes.go
View File

@ -15,14 +15,14 @@ import (
// routesFromSnapshot returns the xDS API representation of the "routes" in the
// snapshot.
func routesFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func routesFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, _ string) ([]proto.Message, error) {
if cfgSnap == nil {
return nil, errors.New("nil config given")
}
switch cfgSnap.Kind {
case structs.ServiceKindConnectProxy:
return routesFromSnapshotConnectProxy(cfgSnap, token)
return routesFromSnapshotConnectProxy(cfgSnap)
default:
return nil, fmt.Errorf("Invalid service kind: %v", cfgSnap.Kind)
}
@ -30,7 +30,7 @@ func routesFromSnapshot(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto
// routesFromSnapshotConnectProxy returns the xDS API representation of the
// "routes" in the snapshot.
func routesFromSnapshotConnectProxy(cfgSnap *proxycfg.ConfigSnapshot, token string) ([]proto.Message, error) {
func routesFromSnapshotConnectProxy(cfgSnap *proxycfg.ConfigSnapshot) ([]proto.Message, error) {
if cfgSnap == nil {
return nil, errors.New("nil config given")
}

21
agent/xds/server.go
View File

@ -190,8 +190,8 @@ func (s *Server) process(stream ADSStream, reqCh <-chan *envoy.DiscoveryRequest)
var nonce uint64
// xDS works with versions of configs. Internally we don't have a consistent
// version. We could just hash the config since versions don't have to be
// ordered as far as I can tell, but it's cheaper just to increment a counter
// version. We could hash the config since versions don't have to be
// ordered as far as I can tell, but it is cheaper to increment a counter
// every time we observe a new config since the upstream proxycfg package only
// delivers updates when there are actual changes.
var configVersion uint64
@ -209,12 +209,12 @@ func (s *Server) process(stream ADSStream, reqCh <-chan *envoy.DiscoveryRequest)
// Configure handlers for each type of request
handlers := map[string]*xDSType{
EndpointType: &xDSType{
EndpointType: {
typeURL: EndpointType,
resources: s.endpointsFromSnapshot,
stream: stream,
},
ClusterType: &xDSType{
ClusterType: {
typeURL: ClusterType,
resources: s.clustersFromSnapshot,
stream: stream,
@ -223,12 +223,12 @@ func (s *Server) process(stream ADSStream, reqCh <-chan *envoy.DiscoveryRequest)
return cfgSnap.Kind == structs.ServiceKindMeshGateway
},
},
RouteType: &xDSType{
RouteType: {
typeURL: RouteType,
resources: routesFromSnapshot,
stream: stream,
},
ListenerType: &xDSType{
ListenerType: {
typeURL: ListenerType,
resources: s.listenersFromSnapshot,
stream: stream,
@ -245,8 +245,7 @@ func (s *Server) process(stream ADSStream, reqCh <-chan *envoy.DiscoveryRequest)
return status.Errorf(codes.Unauthenticated, "unauthenticated: no config snapshot")
}
token := tokenFromStream(stream)
rule, err := s.ResolveToken(token)
rule, err := s.ResolveToken(tokenFromContext(stream.Context()))
if acl.IsErrNotFound(err) {
return status.Errorf(codes.Unauthenticated, "unauthenticated: %v", err)
@ -397,7 +396,7 @@ func (t *xDSType) SendIfNew(cfgSnap *proxycfg.ConfigSnapshot, version uint64, no
// Already sent this version
return nil
}
resources, err := t.resources(cfgSnap, tokenFromStream(t.stream))
resources, err := t.resources(cfgSnap, tokenFromContext(t.stream.Context()))
if err != nil {
return err
}
@ -436,10 +435,6 @@ func (t *xDSType) SendIfNew(cfgSnap *proxycfg.ConfigSnapshot, version uint64, no
return nil
}
func tokenFromStream(stream ADSStream) string {
return tokenFromContext(stream.Context())
}
func tokenFromContext(ctx context.Context) string {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {

2
agent/xds/xds.go
View File

@ -1,4 +1,4 @@
// Package xds provides an impementation of a gRPC service that exports Envoy's
// Package xds provides an implementation of a gRPC service that exports Envoy's
// xDS API for config discovery. Specifically we support the Aggregated
// Discovery Service (ADS) only as we control all config.
//