Skip to content

Commit

Permalink
Fix xds v2 from creating envoy endpoint resources when already inline…
Browse files Browse the repository at this point in the history
…d in the cluster (#19580)

* migrate expose checks and paths  tests to resources_test.go

* fix failing expose paths tests

* fix the way endpoint resources get created to make expose tests pass.

* wip

* remove endpoint resources that are already inlined on local_app clusters

* renaiming and comments
  • Loading branch information
jmurret authored Nov 8, 2023
1 parent 5aff19f commit 7de0b45
Show file tree
Hide file tree
Showing 21 changed files with 96 additions and 639 deletions.
18 changes: 7 additions & 11 deletions agent/xds/resources_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ func TestAllResourcesFromSnapshot(t *testing.T) {
if tt.alsoRunTestForV2 {
generator := xdsv2.NewResourceGenerator(testutil.Logger(t))

converter := proxystateconverter.NewConverter(testutil.Logger(t), &mockCfgFetcher{addressLan: "10.10.10.10"})
converter := proxystateconverter.NewConverter(testutil.Logger(t), &mockCfgFetcher{addressLan: "192.0.2.1"})
proxyState, err := converter.ProxyStateFromSnapshot(snap)
require.NoError(t, err)

Expand Down Expand Up @@ -1057,8 +1057,7 @@ func getExposePathGoldenTestCases() []goldenTestCase {
create: func(t testinf.T) *proxycfg.ConfigSnapshot {
return proxycfg.TestConfigSnapshotExposeConfig(t, nil)
},
// TODO(jm): enable with https://github.com/hashicorp/consul/pull/19459
alsoRunTestForV2: false,
alsoRunTestForV2: true,
},
{
name: "downstream-service-with-unix-sockets",
Expand All @@ -1085,8 +1084,7 @@ func getExposePathGoldenTestCases() []goldenTestCase {
}
})
},
// TODO(jm): enable with https://github.com/hashicorp/consul/pull/19459
alsoRunTestForV2: false,
alsoRunTestForV2: true,
},
{
name: "expose-checks",
Expand All @@ -1096,14 +1094,12 @@ func getExposePathGoldenTestCases() []goldenTestCase {
return "192.0.2.1"
})
},
// TODO(jm): enable with https://github.com/hashicorp/consul/pull/19459
alsoRunTestForV2: false,
alsoRunTestForV2: true,
},
{
name: "expose-paths-grpc-new-cluster-http1",
create: proxycfg.TestConfigSnapshotGRPCExposeHTTP1,
// TODO(jm): enable with https://github.com/hashicorp/consul/pull/19459
alsoRunTestForV2: false,
name: "expose-paths-grpc-new-cluster-http1",
create: proxycfg.TestConfigSnapshotGRPCExposeHTTP1,
alsoRunTestForV2: true,
},
{
// NOTE: if IPv6 is not supported in the kernel per
Expand Down
2 changes: 1 addition & 1 deletion agent/xds/testdata/endpoints/expose-checks.latest.golden
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@
"nonce": "00000001",
"typeUrl": "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment",
"versionInfo": "00000001"
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@
"nonce": "00000001",
"typeUrl": "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment",
"versionInfo": "00000001"
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@
"nonce": "00000001",
"typeUrl": "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment",
"versionInfo": "00000001"
}
}
2 changes: 1 addition & 1 deletion agent/xds/testdata/listeners/expose-checks.latest.golden
Original file line number Diff line number Diff line change
Expand Up @@ -139,4 +139,4 @@
],
"typeUrl": "type.googleapis.com/envoy.config.listener.v3.Listener",
"versionInfo": "00000001"
}
}
2 changes: 1 addition & 1 deletion agent/xds/testdata/routes/expose-checks.latest.golden
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@
"nonce": "00000001",
"typeUrl": "type.googleapis.com/envoy.config.route.v3.RouteConfiguration",
"versionInfo": "00000001"
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@
"nonce": "00000001",
"typeUrl": "type.googleapis.com/envoy.config.route.v3.RouteConfiguration",
"versionInfo": "00000001"
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@
"nonce": "00000001",
"typeUrl": "type.googleapis.com/envoy.config.route.v3.RouteConfiguration",
"versionInfo": "00000001"
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@
"nonce": "00000001",
"typeUrl": "type.googleapis.com/envoy.config.route.v3.RouteConfiguration",
"versionInfo": "00000001"
}
}
109 changes: 75 additions & 34 deletions agent/xdsv2/cluster_resources.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"fmt"
envoy_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
envoy_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
envoy_endpoint_v3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
envoy_aggregate_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/aggregate/v3"
envoy_upstreams_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/upstreams/http/v3"
envoy_type_v3 "github.com/envoyproxy/go-control-plane/envoy/type/v3"
Expand All @@ -18,61 +19,73 @@ import (
"github.com/hashicorp/consul/proto-public/pbmesh/v2beta1/pbproxystate"
)

func (pr *ProxyResources) makeClusters(name string) (map[string]proto.Message, error) {
func (pr *ProxyResources) makeClustersAndEndpoints(name string) (map[string]proto.Message, map[string]proto.Message, error) {
envoyClusters := make(map[string]proto.Message)
envoyEndpoints := make(map[string]proto.Message)
proxyStateCluster, ok := pr.proxyState.Clusters[name]
if !ok {
return nil, fmt.Errorf("cluster %q not found", name)
return nil, nil, fmt.Errorf("cluster %q not found", name)
}

switch proxyStateCluster.Group.(type) {
case *pbproxystate.Cluster_FailoverGroup:
fg := proxyStateCluster.GetFailoverGroup()
clusters, err := pr.makeEnvoyAggregateCluster(name, proxyStateCluster.Protocol, fg)
clusters, eps, err := pr.makeEnvoyAggregateClusterAndEndpoint(name, proxyStateCluster.Protocol, fg)
if err != nil {
return nil, err
return nil, nil, err
}
// for each cluster, add it to clusters map and add endpoint to endpoint map
for _, c := range clusters {
envoyClusters[c.Name] = c
if ep, ok := eps[c.Name]; ok {
envoyEndpoints[c.Name] = ep
}
}

case *pbproxystate.Cluster_EndpointGroup:
eg := proxyStateCluster.GetEndpointGroup()
cluster, err := pr.makeEnvoyCluster(name, proxyStateCluster.Protocol, eg)
cluster, eps, err := pr.makeEnvoyClusterAndEndpoint(name, proxyStateCluster.Protocol, eg)
if err != nil {
return nil, err
return nil, nil, err
}

// for each cluster, add it to clusters map and add endpoint to endpoint map
envoyClusters[cluster.Name] = cluster
if ep, ok := eps[cluster.Name]; ok {
envoyEndpoints[cluster.Name] = ep
}

default:
return nil, errors.New("cluster group type should be Endpoint Group or Failover Group")
return nil, nil, errors.New("cluster group type should be Endpoint Group or Failover Group")
}
return envoyClusters, nil
return envoyClusters, envoyEndpoints, nil
}

func (pr *ProxyResources) makeEnvoyCluster(name string, protocol pbproxystate.Protocol, eg *pbproxystate.EndpointGroup) (*envoy_cluster_v3.Cluster, error) {
func (pr *ProxyResources) makeEnvoyClusterAndEndpoint(name string, protocol pbproxystate.Protocol,
eg *pbproxystate.EndpointGroup) (*envoy_cluster_v3.Cluster, map[string]*envoy_endpoint_v3.ClusterLoadAssignment, error) {
if eg != nil {
switch t := eg.Group.(type) {
case *pbproxystate.EndpointGroup_Dynamic:
dynamic := eg.GetDynamic()
return pr.makeEnvoyDynamicCluster(name, protocol, dynamic)
return pr.makeEnvoyDynamicClusterAndEndpoint(name, protocol, dynamic)
case *pbproxystate.EndpointGroup_Static:
static := eg.GetStatic()
return pr.makeEnvoyStaticCluster(name, protocol, static)
return pr.makeEnvoyStaticClusterAndEndpoint(name, protocol, static)
case *pbproxystate.EndpointGroup_Dns:
dns := eg.GetDns()
return pr.makeEnvoyDnsCluster(name, protocol, dns)
case *pbproxystate.EndpointGroup_Passthrough:
passthrough := eg.GetPassthrough()
return pr.makeEnvoyPassthroughCluster(name, protocol, passthrough)
default:
return nil, fmt.Errorf("unsupported endpoint group type: %s", t)
return nil, nil, fmt.Errorf("unsupported endpoint group type: %s", t)
}
}
return nil, fmt.Errorf("no endpoint group")
return nil, nil, fmt.Errorf("no endpoint group")
}

func (pr *ProxyResources) makeEnvoyDynamicCluster(name string, protocol pbproxystate.Protocol, dynamic *pbproxystate.DynamicEndpointGroup) (*envoy_cluster_v3.Cluster, error) {
func (pr *ProxyResources) makeEnvoyDynamicClusterAndEndpoint(name string, protocol pbproxystate.Protocol,
dynamic *pbproxystate.DynamicEndpointGroup) (*envoy_cluster_v3.Cluster, map[string]*envoy_endpoint_v3.ClusterLoadAssignment, error) {
cluster := &envoy_cluster_v3.Cluster{
Name: name,
ClusterDiscoveryType: &envoy_cluster_v3.Cluster_Type{Type: envoy_cluster_v3.Cluster_EDS},
Expand All @@ -87,7 +100,7 @@ func (pr *ProxyResources) makeEnvoyDynamicCluster(name string, protocol pbproxys
}
err := addHttpProtocolOptions(protocol, cluster)
if err != nil {
return nil, err
return nil, nil, err
}
if dynamic.Config != nil {
if dynamic.Config.UseAltStatName {
Expand All @@ -106,23 +119,31 @@ func (pr *ProxyResources) makeEnvoyDynamicCluster(name string, protocol pbproxys

err := addEnvoyLBToCluster(dynamic.Config, cluster)
if err != nil {
return nil, err
return nil, nil, err
}
}

if dynamic.OutboundTls != nil {
envoyTransportSocket, err := pr.makeEnvoyTransportSocket(dynamic.OutboundTls)
if err != nil {
return nil, err
return nil, nil, err
}
cluster.TransportSocket = envoyTransportSocket
}

return cluster, nil
// Generate Envoy endpoint
endpointResources := make(map[string]*envoy_endpoint_v3.ClusterLoadAssignment)
if endpointList, ok := pr.proxyState.Endpoints[cluster.Name]; ok {
protoEndpoint := makeEnvoyClusterLoadAssignment(cluster.Name, endpointList.Endpoints)
endpointResources[cluster.Name] = protoEndpoint
}

return cluster, endpointResources, nil

}

func (pr *ProxyResources) makeEnvoyStaticCluster(name string, protocol pbproxystate.Protocol, static *pbproxystate.StaticEndpointGroup) (*envoy_cluster_v3.Cluster, error) {
func (pr *ProxyResources) makeEnvoyStaticClusterAndEndpoint(name string, protocol pbproxystate.Protocol,
static *pbproxystate.StaticEndpointGroup) (*envoy_cluster_v3.Cluster, map[string]*envoy_endpoint_v3.ClusterLoadAssignment, error) {
cluster := &envoy_cluster_v3.Cluster{
Name: name,
ClusterDiscoveryType: &envoy_cluster_v3.Cluster_Type{Type: envoy_cluster_v3.Cluster_STATIC},
Expand All @@ -141,21 +162,23 @@ func (pr *ProxyResources) makeEnvoyStaticCluster(name string, protocol pbproxyst
err = addHttpProtocolOptions(protocol, cluster)
}
if err != nil {
return nil, err
return nil, nil, err
}

if static.Config != nil {
cluster.ConnectTimeout = static.Config.ConnectTimeout
addEnvoyCircuitBreakers(static.GetConfig().CircuitBreakers, cluster)
}
return cluster, nil
return cluster, nil, nil
}

func (pr *ProxyResources) makeEnvoyDnsCluster(name string, protocol pbproxystate.Protocol, dns *pbproxystate.DNSEndpointGroup) (*envoy_cluster_v3.Cluster, error) {
return nil, nil
func (pr *ProxyResources) makeEnvoyDnsCluster(name string, protocol pbproxystate.Protocol,
dns *pbproxystate.DNSEndpointGroup) (*envoy_cluster_v3.Cluster, map[string]*envoy_endpoint_v3.ClusterLoadAssignment, error) {
return nil, nil, nil
}

func (pr *ProxyResources) makeEnvoyPassthroughCluster(name string, protocol pbproxystate.Protocol, passthrough *pbproxystate.PassthroughEndpointGroup) (*envoy_cluster_v3.Cluster, error) {
func (pr *ProxyResources) makeEnvoyPassthroughCluster(name string, protocol pbproxystate.Protocol,
passthrough *pbproxystate.PassthroughEndpointGroup) (*envoy_cluster_v3.Cluster, map[string]*envoy_endpoint_v3.ClusterLoadAssignment, error) {
cluster := &envoy_cluster_v3.Cluster{
Name: name,
ConnectTimeout: passthrough.Config.ConnectTimeout,
Expand All @@ -165,37 +188,47 @@ func (pr *ProxyResources) makeEnvoyPassthroughCluster(name string, protocol pbpr
if passthrough.OutboundTls != nil {
envoyTransportSocket, err := pr.makeEnvoyTransportSocket(passthrough.OutboundTls)
if err != nil {
return nil, err
return nil, nil, err
}
cluster.TransportSocket = envoyTransportSocket
}
err := addHttpProtocolOptions(protocol, cluster)
if err != nil {
return nil, err
return nil, nil, err
}
return cluster, nil
return cluster, nil, nil
}

func (pr *ProxyResources) makeEnvoyAggregateCluster(name string, protocol pbproxystate.Protocol, fg *pbproxystate.FailoverGroup) (map[string]*envoy_cluster_v3.Cluster, error) {
func (pr *ProxyResources) makeEnvoyAggregateClusterAndEndpoint(name string, protocol pbproxystate.Protocol,
fg *pbproxystate.FailoverGroup) (map[string]*envoy_cluster_v3.Cluster, map[string]*envoy_endpoint_v3.ClusterLoadAssignment, error) {
clusters := make(map[string]*envoy_cluster_v3.Cluster)
endpointResources := make(map[string]*envoy_endpoint_v3.ClusterLoadAssignment)
if fg != nil {
var egNames []string
for _, eg := range fg.EndpointGroups {
cluster, err := pr.makeEnvoyCluster(eg.Name, protocol, eg)
cluster, eps, err := pr.makeEnvoyClusterAndEndpoint(eg.Name, protocol, eg)
if err != nil {
return nil, err
return nil, eps, err
}
egNames = append(egNames, cluster.Name)

// add failover cluster
clusters[cluster.Name] = cluster

// add endpoint for failover cluster
if ep, ok := eps[cluster.Name]; ok {
endpointResources[cluster.Name] = ep
}
}
aggregateClusterConfig, err := anypb.New(&envoy_aggregate_cluster_v3.ClusterConfig{
Clusters: egNames,
})

if err != nil {
return nil, err
return nil, nil, err
}

// create aggregate cluster
c := &envoy_cluster_v3.Cluster{
Name: name,
ConnectTimeout: fg.Config.ConnectTimeout,
Expand All @@ -212,11 +245,19 @@ func (pr *ProxyResources) makeEnvoyAggregateCluster(name string, protocol pbprox
}
err = addHttpProtocolOptions(protocol, c)
if err != nil {
return nil, err
return nil, nil, err
}

// add aggregate cluster
clusters[c.Name] = c

// add endpoint for aggregate cluster
if endpointList, ok := pr.proxyState.Endpoints[c.Name]; ok {
protoEndpoint := makeEnvoyClusterLoadAssignment(c.Name, endpointList.Endpoints)
endpointResources[c.Name] = protoEndpoint
}
}
return clusters, nil
return clusters, endpointResources, nil
}

func addLocalAppHttpProtocolOptions(protocol pbproxystate.Protocol, c *envoy_cluster_v3.Cluster) error {
Expand Down Expand Up @@ -345,7 +386,7 @@ func addEnvoyLBToCluster(dynamicConfig *pbproxystate.DynamicEndpointGroupConfig,
return nil
}

func (pr *ProxyResources) makeEnvoyClusterFromL4Destination(l4 *pbproxystate.L4Destination) error {
func (pr *ProxyResources) makeEnvoyClustersAndEndpointsFromL4Destination(l4 *pbproxystate.L4Destination) error {
switch l4.Destination.(type) {
case *pbproxystate.L4Destination_Cluster:
pr.addEnvoyClustersAndEndpointsToEnvoyResources(l4.GetCluster().GetName())
Expand Down
2 changes: 1 addition & 1 deletion agent/xdsv2/listener_resources.go
Original file line number Diff line number Diff line change
Expand Up @@ -303,7 +303,7 @@ func (pr *ProxyResources) makeEnvoyResourcesForSNIDestination(sni *pbproxystate.
}

func (pr *ProxyResources) makeEnvoyResourcesForL4Destination(l4 *pbproxystate.Router_L4) ([]*envoy_listener_v3.Filter, error) {
err := pr.makeEnvoyClusterFromL4Destination(l4.L4)
err := pr.makeEnvoyClustersAndEndpointsFromL4Destination(l4.L4)
if err != nil {
return nil, err
}
Expand Down
12 changes: 5 additions & 7 deletions agent/xdsv2/route_resources.go
Original file line number Diff line number Diff line change
Expand Up @@ -235,16 +235,14 @@ func makeEnvoyQueryParamFromProxystateQueryMatch(psMatch *pbproxystate.QueryPara
}

func (pr *ProxyResources) addEnvoyClustersAndEndpointsToEnvoyResources(clusterName string) {
clusters, _ := pr.makeClusters(clusterName)
clusters, endpoints, _ := pr.makeClustersAndEndpoints(clusterName)

for name, cluster := range clusters {
pr.envoyResources[xdscommon.ClusterType][name] = cluster
}

if name != xdscommon.LocalAppClusterName {
if endpointList, ok := pr.proxyState.Endpoints[name]; ok {
protoEndpoint := makeEnvoyClusterLoadAssignment(name, endpointList.Endpoints)
pr.envoyResources[xdscommon.EndpointType][name] = protoEndpoint
}
}
for name, ep := range endpoints {
pr.envoyResources[xdscommon.EndpointType][name] = ep
}
}

Expand Down
Loading

0 comments on commit 7de0b45

Please sign in to comment.