From 6c0b842528d80ad8b4704828cc0e0601e1094952 Mon Sep 17 00:00:00 2001 From: Mike Beaumont Date: Tue, 20 Dec 2022 13:23:02 +0100 Subject: [PATCH] feat(MeshHTTPRoute): initial plugin Signed-off-by: Mike Beaumont --- .../api/v1alpha1/meshhttproute.go | 5 + .../policies/meshhttproute/common/types.go | 12 ++ .../meshhttproute/plugin/v1alpha1/clusters.go | 99 +++++++++++ .../plugin/v1alpha1/endpoints.go | 48 +++++ .../plugin/v1alpha1/listeners.go | 156 ++++++++++++++++ .../meshhttproute/plugin/v1alpha1/plugin.go | 167 ++++++++++++++++++ .../policies/meshhttproute/xds/builder.go | 42 +++++ .../policies/meshhttproute/xds/configurer.go | 86 +++++++++ pkg/xds/generator/outbound_proxy_generator.go | 8 +- 9 files changed, 619 insertions(+), 4 deletions(-) create mode 100644 pkg/plugins/policies/meshhttproute/common/types.go create mode 100644 pkg/plugins/policies/meshhttproute/plugin/v1alpha1/clusters.go create mode 100644 pkg/plugins/policies/meshhttproute/plugin/v1alpha1/endpoints.go create mode 100644 pkg/plugins/policies/meshhttproute/plugin/v1alpha1/listeners.go create mode 100644 pkg/plugins/policies/meshhttproute/plugin/v1alpha1/plugin.go create mode 100644 pkg/plugins/policies/meshhttproute/xds/builder.go create mode 100644 pkg/plugins/policies/meshhttproute/xds/configurer.go diff --git a/pkg/plugins/policies/meshhttproute/api/v1alpha1/meshhttproute.go b/pkg/plugins/policies/meshhttproute/api/v1alpha1/meshhttproute.go index c40690b4f4e9..696a67f186f9 100644 --- a/pkg/plugins/policies/meshhttproute/api/v1alpha1/meshhttproute.go +++ b/pkg/plugins/policies/meshhttproute/api/v1alpha1/meshhttproute.go @@ -7,6 +7,11 @@ import ( // MeshHTTPRoute // +kuma:policy:skip_registration=true +// +// This policy defines its own `GetDefault` method so that it can have the given +// structure for deserialization but still use the generic policy merging +// machinery. +// // +kuma:policy:skip_get_default=true type MeshHTTPRoute struct { // TargetRef is a reference to the resource the policy takes an effect on. diff --git a/pkg/plugins/policies/meshhttproute/common/types.go b/pkg/plugins/policies/meshhttproute/common/types.go new file mode 100644 index 000000000000..56acb88e199c --- /dev/null +++ b/pkg/plugins/policies/meshhttproute/common/types.go @@ -0,0 +1,12 @@ +package common + +import ( + api "github.com/kumahq/kuma/pkg/plugins/policies/meshhttproute/api/v1alpha1" + envoy_common "github.com/kumahq/kuma/pkg/xds/envoy" +) + +type Route struct { + Matches []api.Match + Filters []api.Filter + BackendRefs []envoy_common.Cluster +} diff --git a/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/clusters.go b/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/clusters.go new file mode 100644 index 000000000000..c55b85ae62d3 --- /dev/null +++ b/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/clusters.go @@ -0,0 +1,99 @@ +package v1alpha1 + +import ( + "github.com/pkg/errors" + + mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1" + core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh" + core_xds "github.com/kumahq/kuma/pkg/core/xds" + xds_context "github.com/kumahq/kuma/pkg/xds/context" + envoy_common "github.com/kumahq/kuma/pkg/xds/envoy" + envoy_clusters "github.com/kumahq/kuma/pkg/xds/envoy/clusters" + envoy_tags "github.com/kumahq/kuma/pkg/xds/envoy/tags" + "github.com/kumahq/kuma/pkg/xds/generator" +) + +func generateClusters( + proxy *core_xds.Proxy, + meshCtx xds_context.MeshContext, + services envoy_common.Services, +) (*core_xds.ResourceSet, error) { + resources := core_xds.NewResourceSet() + + for _, serviceName := range services.Sorted() { + service := services[serviceName] + protocol := generator.InferProtocol(proxy, service.Clusters()) + tlsReady := service.TLSReady() + + for _, cluster := range service.Clusters() { + edsClusterBuilder := envoy_clusters.NewClusterBuilder(proxy.APIVersion) + + clusterName := cluster.Name() + clusterTags := []envoy_tags.Tags{cluster.Tags()} + + if service.HasExternalService() { + if meshCtx.Resource.ZoneEgressEnabled() { + edsClusterBuilder. + Configure(envoy_clusters.EdsCluster(clusterName)). + Configure(envoy_clusters.ClientSideMTLS( + proxy.SecretsTracker, + meshCtx.Resource, + mesh_proto.ZoneEgressServiceName, + tlsReady, + clusterTags, + )) + } else { + endpoints := proxy.Routing.ExternalServiceOutboundTargets[serviceName] + isIPv6 := proxy.Dataplane.IsIPv6() + + edsClusterBuilder. + Configure(envoy_clusters.ProvidedEndpointCluster(clusterName, isIPv6, endpoints...)). + Configure(envoy_clusters.ClientSideTLS(endpoints)) + } + + switch protocol { + case core_mesh.ProtocolHTTP: + edsClusterBuilder.Configure(envoy_clusters.Http()) + case core_mesh.ProtocolHTTP2, core_mesh.ProtocolGRPC: + edsClusterBuilder.Configure(envoy_clusters.Http2()) + default: + } + } else { + edsClusterBuilder. + Configure(envoy_clusters.EdsCluster(clusterName)). + Configure(envoy_clusters.LB(cluster.LB())). + Configure(envoy_clusters.Http2()) + + if upstreamMeshName := cluster.Mesh(); upstreamMeshName != "" { + for _, otherMesh := range append(meshCtx.Resources.OtherMeshes().Items, meshCtx.Resource) { + if otherMesh.GetMeta().GetName() == upstreamMeshName { + edsClusterBuilder.Configure( + envoy_clusters.CrossMeshClientSideMTLS( + proxy.SecretsTracker, meshCtx.Resource, otherMesh, serviceName, tlsReady, clusterTags, + ), + ) + break + } + } + } else { + edsClusterBuilder.Configure(envoy_clusters.ClientSideMTLS( + proxy.SecretsTracker, + meshCtx.Resource, serviceName, tlsReady, clusterTags)) + } + } + + edsCluster, err := edsClusterBuilder.Build() + if err != nil { + return nil, errors.Wrapf(err, "build CDS for cluster %s failed", clusterName) + } + + resources = resources.Add(&core_xds.Resource{ + Name: clusterName, + Origin: generator.OriginOutbound, + Resource: edsCluster, + }) + } + } + + return resources, nil +} diff --git a/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/endpoints.go b/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/endpoints.go new file mode 100644 index 000000000000..4c7bf33e469f --- /dev/null +++ b/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/endpoints.go @@ -0,0 +1,48 @@ +package v1alpha1 + +import ( + "context" + + "github.com/pkg/errors" + + "github.com/kumahq/kuma/pkg/core/user" + core_xds "github.com/kumahq/kuma/pkg/core/xds" + xds_context "github.com/kumahq/kuma/pkg/xds/context" + envoy_common "github.com/kumahq/kuma/pkg/xds/envoy" +) + +func generateEndpoints( + proxy *core_xds.Proxy, + ctx xds_context.Context, + services envoy_common.Services, +) (*core_xds.ResourceSet, error) { + resources := core_xds.NewResourceSet() + + for _, serviceName := range services.Sorted() { + // When no zone egress is present in a mesh Endpoints for ExternalServices + // are specified in load assignment in DNS Cluster. + // We are not allowed to add endpoints with DNS names through EDS. + if !services[serviceName].HasExternalService() || ctx.Mesh.Resource.ZoneEgressEnabled() { + for _, cluster := range services[serviceName].Clusters() { + var endpoints core_xds.EndpointMap + if cluster.Mesh() != "" { + endpoints = ctx.Mesh.CrossMeshEndpoints[cluster.Mesh()] + } else { + endpoints = ctx.Mesh.EndpointMap + } + + loadAssignment, err := ctx.ControlPlane.CLACache.GetCLA(user.Ctx(context.TODO(), user.ControlPlane), proxy.Dataplane.GetMeta().GetMesh(), ctx.Mesh.Hash, cluster, proxy.APIVersion, endpoints) + if err != nil { + return nil, errors.Wrapf(err, "could not get ClusterLoadAssignment for %s", serviceName) + } + + resources.Add(&core_xds.Resource{ + Name: cluster.Name(), + Resource: loadAssignment, + }) + } + } + } + + return resources, nil +} diff --git a/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/listeners.go b/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/listeners.go new file mode 100644 index 000000000000..ef2447bae6da --- /dev/null +++ b/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/listeners.go @@ -0,0 +1,156 @@ +package v1alpha1 + +import ( + "fmt" + + common_api "github.com/kumahq/kuma/api/common/v1alpha1" + mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1" + core_xds "github.com/kumahq/kuma/pkg/core/xds" + api "github.com/kumahq/kuma/pkg/plugins/policies/meshhttproute/api/v1alpha1" + xds "github.com/kumahq/kuma/pkg/plugins/policies/meshhttproute/xds" + "github.com/kumahq/kuma/pkg/xds/envoy" + envoy_common "github.com/kumahq/kuma/pkg/xds/envoy" + envoy_listeners "github.com/kumahq/kuma/pkg/xds/envoy/listeners" + envoy_names "github.com/kumahq/kuma/pkg/xds/envoy/names" + envoy_tags "github.com/kumahq/kuma/pkg/xds/envoy/tags" + "github.com/kumahq/kuma/pkg/xds/generator" +) + +func generateListeners( + proxy *core_xds.Proxy, + rules []ToRouteRule, + servicesAcc envoy_common.ServicesAccumulator, +) (*core_xds.ResourceSet, error) { + resources := core_xds.NewResourceSet() + splitCounter := &splitCounter{} + + for _, outbound := range proxy.Dataplane.Spec.GetNetworking().GetOutbound() { + serviceName := outbound.GetTagsIncludingLegacy()[mesh_proto.ServiceTag] + oface := proxy.Dataplane.Spec.Networking.ToOutboundInterface(outbound) + outboundListenerName := envoy_names.GetOutboundListenerName(oface.DataplaneIP, oface.DataplanePort) + + listenerBuilder := envoy_listeners.NewListenerBuilder(proxy.APIVersion). + Configure(envoy_listeners.OutboundListener(outboundListenerName, oface.DataplaneIP, oface.DataplanePort, core_xds.SocketAddressProtocolTCP)). + Configure(envoy_listeners.TransparentProxying(proxy.Dataplane.Spec.Networking.GetTransparentProxying())). + Configure(envoy_listeners.TagsMetadata(envoy_tags.Tags(outbound.GetTagsIncludingLegacy()).WithoutTags(mesh_proto.MeshTag))) + + for _, route := range FindRoutes(rules, serviceName) { + clusters := makeClusters(proxy, map[string]string{}, splitCounter, route.BackendRefs) + servicesAcc.Add(clusters...) + + filterChainBuilder := envoy_listeners.NewFilterChainBuilder(proxy.APIVersion) + + serviceName := outbound.GetTagsIncludingLegacy()[mesh_proto.ServiceTag] + configurer := &xds.HttpOutboundRouteConfigurer{ + Service: serviceName, + Matches: route.Matches, + Filters: route.Filters, + Clusters: clusters, + DpTags: proxy.Dataplane.Spec.TagSet(), + } + filterChainBuilder. + Configure(envoy_listeners.HttpConnectionManager(serviceName, false)). + Configure(envoy_listeners.AddFilterChainConfigurer(configurer)) + + listenerBuilder = listenerBuilder.Configure(envoy_listeners.FilterChain(filterChainBuilder)) + } + + listener, err := listenerBuilder.Build() + if err != nil { + return nil, err + } + resources.Add(&core_xds.Resource{ + Name: listener.GetName(), + Origin: generator.OriginOutbound, + Resource: listener, + }) + } + + return resources, nil +} + +// Whenever `split` is specified in the TrafficRoute which has more than kuma.io/service tag +// We generate a separate Envoy cluster with _X_ suffix. SplitCounter ensures that we have different X for every split in one Dataplane +// Each split is distinct for the whole Dataplane so we can avoid accidental cluster overrides. +type splitCounter struct { + counter int +} + +func (s *splitCounter) getAndIncrement() int { + counter := s.counter + s.counter++ + return counter +} + +func makeClusters( + proxy *core_xds.Proxy, + clusterCache map[string]string, + splitCounter *splitCounter, + refs []api.BackendRef, +) []envoy.Cluster { + var clustersInternal, clustersExternal []envoy.Cluster + + for _, ref := range refs { + switch ref.Kind { + case common_api.MeshService, common_api.MeshServiceSubset: + default: + continue + } + + service := ref.Name + if ref.Weight == 0 { + continue + } + + name := service + + if len(ref.Tags) > 0 { + name = envoy_names.GetSplitClusterName(service, splitCounter.getAndIncrement()) + } + + if mesh, ok := ref.Tags[mesh_proto.MeshTag]; ok { + // The name should be distinct to the service & mesh combination + name = fmt.Sprintf("%s_%s", name, mesh) + } + + // We assume that all the targets are either ExternalServices or not + // therefore we check only the first one + var isExternalService bool + if endpoints := proxy.Routing.OutboundTargets[service]; len(endpoints) > 0 { + isExternalService = endpoints[0].IsExternalService() + } + if endpoints := proxy.Routing.ExternalServiceOutboundTargets[service]; len(endpoints) > 0 { + isExternalService = true + } + + allTags := envoy_tags.Tags(ref.Tags).WithTags(mesh_proto.ServiceTag, ref.Name) + cluster := envoy_common.NewCluster( + envoy_common.WithService(service), + envoy_common.WithName(name), + envoy_common.WithWeight(uint32(ref.Weight)), + // The mesh tag is set here if this destination is generated + // from a MeshGateway virtual outbound and is not part of the + // service tags + envoy_common.WithTags(allTags.WithoutTags(mesh_proto.MeshTag)), + envoy_common.WithExternalService(isExternalService), + ) + + if mesh, ok := ref.Tags[mesh_proto.MeshTag]; ok { + cluster.SetMesh(mesh) + } + + if name, ok := clusterCache[allTags.String()]; ok { + cluster.SetName(name) + } else { + clusterCache[allTags.String()] = cluster.Name() + } + + if isExternalService { + clustersExternal = append(clustersExternal, cluster) + } else { + clustersInternal = append(clustersInternal, cluster) + } + } + + return append(clustersExternal, clustersInternal...) +} diff --git a/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/plugin.go b/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/plugin.go new file mode 100644 index 000000000000..94a2163860e7 --- /dev/null +++ b/pkg/plugins/policies/meshhttproute/plugin/v1alpha1/plugin.go @@ -0,0 +1,167 @@ +package v1alpha1 + +import ( + "reflect" + + "github.com/pkg/errors" + + common_api "github.com/kumahq/kuma/api/common/v1alpha1" + core_plugins "github.com/kumahq/kuma/pkg/core/plugins" + core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh" + core_model "github.com/kumahq/kuma/pkg/core/resources/model" + core_xds "github.com/kumahq/kuma/pkg/core/xds" + "github.com/kumahq/kuma/pkg/plugins/policies/matchers" + api "github.com/kumahq/kuma/pkg/plugins/policies/meshhttproute/api/v1alpha1" + xds_context "github.com/kumahq/kuma/pkg/xds/context" + envoy_common "github.com/kumahq/kuma/pkg/xds/envoy" +) + +var _ core_plugins.PolicyPlugin = &plugin{} + +type Route struct { + Matches []api.Match + Filters []api.Filter + BackendRefs []api.BackendRef +} + +type RuleAcc struct { + MatchKey []api.Match + RuleConfs []api.RuleConf +} + +type ToRouteRule struct { + Subset core_xds.Subset + Rules []api.Rule + Origin []core_model.ResourceMeta +} + +type plugin struct { +} + +func NewPlugin() core_plugins.Plugin { + return &plugin{} +} + +func (p plugin) MatchedPolicies(dataplane *core_mesh.DataplaneResource, resources xds_context.Resources) (core_xds.TypedMatchingPolicies, error) { + return matchers.MatchedPolicies(api.MeshHTTPRouteType, dataplane, resources) +} + +func (p plugin) Apply(rs *core_xds.ResourceSet, ctx xds_context.Context, proxy *core_xds.Proxy) error { + // These policies have alreadu been merged using the custom `GetDefault` + // method and therefore are of the + // `ToRouteRule` type, where rules have been appended together. + policies := proxy.Policies.Dynamic[api.MeshHTTPRouteType] + + var toRules []ToRouteRule + for _, policy := range policies.ToRules.Rules { + toRules = append(toRules, ToRouteRule{ + Subset: policy.Subset, + Rules: policy.Conf.([]api.Rule), + Origin: policy.Origin, + }) + } + + if err := ApplyToOutbounds(proxy, rs, ctx, toRules); err != nil { + return err + } + return nil +} + +func ApplyToOutbounds( + proxy *core_xds.Proxy, + rs *core_xds.ResourceSet, + ctx xds_context.Context, + rules []ToRouteRule, +) error { + servicesAcc := envoy_common.NewServicesAccumulator(ctx.Mesh.ServiceTLSReadiness) + + listeners, err := generateListeners(proxy, rules, servicesAcc) + if err != nil { + return errors.Wrap(err, "couldn't generate listener resources") + } + rs.AddSet(listeners) + + services := servicesAcc.Services() + + clusters, err := generateClusters(proxy, ctx.Mesh, services) + if err != nil { + return errors.Wrap(err, "couldn't generate cluster resources") + } + rs.AddSet(clusters) + + endpoints, err := generateEndpoints(proxy, ctx, services) + if err != nil { + return errors.Wrap(err, "couldn't generate endpoint resources") + } + rs.AddSet(endpoints) + + return nil +} + +func FindRoutes( + rules []ToRouteRule, + serviceName string, +) []Route { + var unmergedRules []RuleAcc + + for _, rule := range rules { + if !rule.Subset.IsSubset(core_xds.MeshService(serviceName)) { + continue + } + for _, routeRules := range rule.Rules { + var found bool + for i, accRule := range unmergedRules { + if !reflect.DeepEqual(accRule.MatchKey, routeRules.Matches) { + continue + } + unmergedRules[i] = RuleAcc{ + MatchKey: accRule.MatchKey, + RuleConfs: append(accRule.RuleConfs, routeRules.Default), + } + found = true + } + if !found { + unmergedRules = append(unmergedRules, RuleAcc{ + MatchKey: routeRules.Matches, + RuleConfs: []api.RuleConf{routeRules.Default}, + }) + } + } + } + + var routes []Route + + for _, rule := range unmergedRules { + route := Route{ + Matches: rule.MatchKey, + } + for _, conf := range rule.RuleConfs { + if conf.Filters != nil { + route.Filters = *conf.Filters + } + if conf.BackendRefs != nil { + route.BackendRefs = *conf.BackendRefs + } + } + routes = append(routes, route) + } + + // append the default route + routes = append(routes, Route{ + Matches: []api.Match{{ + Path: api.PathMatch{ + Prefix: "/", + }, + }}, + Filters: nil, + BackendRefs: []api.BackendRef{{ + TargetRef: common_api.TargetRef{ + Kind: common_api.MeshService, + Name: serviceName, + }, + Weight: 100, + }}, + }) + + return routes +} diff --git a/pkg/plugins/policies/meshhttproute/xds/builder.go b/pkg/plugins/policies/meshhttproute/xds/builder.go new file mode 100644 index 000000000000..688d0c1b95b4 --- /dev/null +++ b/pkg/plugins/policies/meshhttproute/xds/builder.go @@ -0,0 +1,42 @@ +package xds + +import ( + envoy_listener "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3" + + mesh_proto "github.com/kumahq/kuma/api/mesh/v1alpha1" + api "github.com/kumahq/kuma/pkg/plugins/policies/meshhttproute/api/v1alpha1" + envoy_common "github.com/kumahq/kuma/pkg/xds/envoy" + envoy_listeners_v3 "github.com/kumahq/kuma/pkg/xds/envoy/listeners/v3" + envoy_names "github.com/kumahq/kuma/pkg/xds/envoy/names" + envoy_routes "github.com/kumahq/kuma/pkg/xds/envoy/routes" +) + +type HttpOutboundRouteConfigurer struct { + Service string + Matches []api.Match + Filters []api.Filter + Clusters []envoy_common.Cluster + DpTags mesh_proto.MultiValueTagSet +} + +var _ envoy_listeners_v3.FilterChainConfigurer = &HttpOutboundRouteConfigurer{} + +func (c *HttpOutboundRouteConfigurer) Configure(filterChain *envoy_listener.FilterChain) error { + routes := envoy_routes.AddVirtualHostConfigurer( + &RoutesConfigurer{ + Matches: c.Matches, + Filters: c.Filters, + Clusters: c.Clusters, + }) + + static := envoy_listeners_v3.HttpStaticRouteConfigurer{ + Builder: envoy_routes.NewRouteConfigurationBuilder(envoy_common.APIV3). + Configure(envoy_routes.CommonRouteConfiguration(envoy_names.GetOutboundRouteName(c.Service))). + Configure(envoy_routes.TagsHeader(c.DpTags)). + Configure(envoy_routes.VirtualHost(envoy_routes.NewVirtualHostBuilder(envoy_common.APIV3). + Configure(envoy_routes.CommonVirtualHost(c.Service)). + Configure(routes))), + } + + return static.Configure(filterChain) +} diff --git a/pkg/plugins/policies/meshhttproute/xds/configurer.go b/pkg/plugins/policies/meshhttproute/xds/configurer.go new file mode 100644 index 000000000000..7846897eaf3b --- /dev/null +++ b/pkg/plugins/policies/meshhttproute/xds/configurer.go @@ -0,0 +1,86 @@ +package xds + +import ( + envoy_route "github.com/envoyproxy/go-control-plane/envoy/config/route/v3" + "google.golang.org/protobuf/types/known/anypb" + + api "github.com/kumahq/kuma/pkg/plugins/policies/meshhttproute/api/v1alpha1" + util_proto "github.com/kumahq/kuma/pkg/util/proto" + envoy_common "github.com/kumahq/kuma/pkg/xds/envoy" +) + +type RoutesConfigurer struct { + Matches []api.Match + Filters []api.Filter + Clusters []envoy_common.Cluster +} + +func (c RoutesConfigurer) Configure(virtualHost *envoy_route.VirtualHost) error { + envoyRoute := &envoy_route.Route{ + Match: c.routeMatch(c.Matches), + Action: &envoy_route.Route_Route{ + Route: c.routeAction(c.Clusters, c.Filters), + }, + TypedPerFilterConfig: map[string]*anypb.Any{}, + } + + virtualHost.Routes = append(virtualHost.Routes, envoyRoute) + return nil +} + +func (c RoutesConfigurer) routeMatch(matches []api.Match) *envoy_route.RouteMatch { + envoyMatch := &envoy_route.RouteMatch{} + + for _, match := range matches { + if match.Path.Prefix != "" { + envoyMatch.PathSpecifier = &envoy_route.RouteMatch_Prefix{ + Prefix: match.Path.Prefix, + } + } + } + + return envoyMatch +} + +func (c RoutesConfigurer) hasExternal(clusters []envoy_common.Cluster) bool { + for _, cluster := range clusters { + if cluster.IsExternalService() { + return true + } + } + return false +} + +func (c RoutesConfigurer) routeAction(clusters []envoy_common.Cluster, _ []api.Filter) *envoy_route.RouteAction { + routeAction := &envoy_route.RouteAction{} + if len(clusters) != 0 { + routeAction.Timeout = util_proto.Duration(clusters[0].Timeout().GetHttp().GetRequestTimeout().AsDuration()) + } + if len(clusters) == 1 { + routeAction.ClusterSpecifier = &envoy_route.RouteAction_Cluster{ + Cluster: clusters[0].Name(), + } + } else { + var weightedClusters []*envoy_route.WeightedCluster_ClusterWeight + var totalWeight uint32 + for _, cluster := range clusters { + weightedClusters = append(weightedClusters, &envoy_route.WeightedCluster_ClusterWeight{ + Name: cluster.Name(), + Weight: util_proto.UInt32(cluster.Weight()), + }) + totalWeight += cluster.Weight() + } + routeAction.ClusterSpecifier = &envoy_route.RouteAction_WeightedClusters{ + WeightedClusters: &envoy_route.WeightedCluster{ + Clusters: weightedClusters, + TotalWeight: util_proto.UInt32(totalWeight), + }, + } + } + if c.hasExternal(clusters) { + routeAction.HostRewriteSpecifier = &envoy_route.RouteAction_AutoHostRewrite{ + AutoHostRewrite: util_proto.Bool(true), + } + } + return routeAction +} diff --git a/pkg/xds/generator/outbound_proxy_generator.go b/pkg/xds/generator/outbound_proxy_generator.go index f83e058e47f7..4ec958ef189f 100644 --- a/pkg/xds/generator/outbound_proxy_generator.go +++ b/pkg/xds/generator/outbound_proxy_generator.go @@ -62,7 +62,7 @@ func (g OutboundProxyGenerator) Generate(ctx xds_context.Context, proxy *model.P clusters := routes.Clusters() servicesAcc.Add(clusters...) - protocol := g.inferProtocol(proxy, clusters) + protocol := InferProtocol(proxy, clusters) // Generate listener listener, err := g.generateLDS(ctx, proxy, routes, outbound, protocol) @@ -193,7 +193,7 @@ func (g OutboundProxyGenerator) generateCDS(ctx xds_context.Context, services en service := services[serviceName] healthCheck := proxy.Policies.HealthChecks[serviceName] circuitBreaker := proxy.Policies.CircuitBreakers[serviceName] - protocol := g.inferProtocol(proxy, service.Clusters()) + protocol := InferProtocol(proxy, service.Clusters()) tlsReady := service.TLSReady() for _, cluster := range service.Clusters() { @@ -310,8 +310,8 @@ func (OutboundProxyGenerator) generateEDS( return resources, nil } -// inferProtocol infers protocol for the destination listener. It will only return HTTP when all endpoints are tagged with HTTP. -func (OutboundProxyGenerator) inferProtocol(proxy *model.Proxy, clusters []envoy_common.Cluster) core_mesh.Protocol { +// InferProtocol infers protocol for the destination listener. It will only return HTTP when all endpoints are tagged with HTTP. +func InferProtocol(proxy *model.Proxy, clusters []envoy_common.Cluster) core_mesh.Protocol { var allEndpoints []model.Endpoint for _, cluster := range clusters { serviceName := cluster.Tags()[mesh_proto.ServiceTag]