422 lines
14 KiB
Go
422 lines
14 KiB
Go
// Copyright (c) HashiCorp, Inc.
|
|
// SPDX-License-Identifier: MPL-2.0
|
|
|
|
package validate
|
|
|
|
import (
|
|
"fmt"
|
|
"strings"
|
|
|
|
envoy_admin_v3 "github.com/envoyproxy/go-control-plane/envoy/admin/v3"
|
|
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_listener_v3 "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
|
|
envoy_route_v3 "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
|
|
envoy_aggregate_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/aggregate/v3"
|
|
envoy_resource_v3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
|
|
"github.com/hashicorp/consul/api"
|
|
"google.golang.org/protobuf/proto"
|
|
"google.golang.org/protobuf/types/known/anypb"
|
|
|
|
"github.com/hashicorp/consul/envoyextensions/extensioncommon"
|
|
)
|
|
|
|
// Validate contains input information about which proxy resources to validate and output information about resources it
|
|
// has validated.
|
|
type Validate struct {
|
|
extensioncommon.BasicExtensionAdapter
|
|
|
|
// envoyID is an argument to the Validate plugin and identifies which listener to begin the validation with.
|
|
envoyID string
|
|
|
|
// vip is an argument to the Validate plugin and identifies which transparent proxy listener to begin the validation
|
|
// with.
|
|
vip string
|
|
|
|
// snis is all of the upstream SNIs for this proxy. It is set via ExtensionConfiguration.
|
|
snis map[string]struct{}
|
|
|
|
// listener specifies if the service's listener has been seen.
|
|
listener bool
|
|
|
|
// usesRDS determines if the listener's outgoing filter uses RDS.
|
|
usesRDS bool
|
|
|
|
// listener specifies if the service's route has been seen.
|
|
route bool
|
|
|
|
// resources is a mapping from SNI to the expected resources
|
|
// for that SNI. It is populated based on the cluster names on routes
|
|
// (whether they are specified on listener filters or routes).
|
|
resources map[string]*resource
|
|
}
|
|
|
|
type resource struct {
|
|
// required determines if the resource is required for the given upstream.
|
|
required bool
|
|
// cluster specifies if the cluster has been seen.
|
|
cluster bool
|
|
// aggregateCluster determines if the resource is an aggregate cluster.
|
|
aggregateCluster bool
|
|
// aggregateClusterChildren is a list of SNIs to identify the child clusters of this aggregate cluster.
|
|
aggregateClusterChildren []string
|
|
// parentCluster is empty if this is a top level cluster, and has a value if this is a child of an aggregate
|
|
// cluster.
|
|
parentCluster string
|
|
// loadAssignment specifies if the load assignment has been seen.
|
|
loadAssignment bool
|
|
// usesEDS specifies if the cluster has EDS configured.
|
|
usesEDS bool
|
|
// The number of endpoints for the cluster or load assignment.
|
|
endpoints int
|
|
}
|
|
|
|
var _ extensioncommon.BasicExtension = (*Validate)(nil)
|
|
|
|
// EndpointValidator allows us to inject a different function for tests.
|
|
type EndpointValidator func(*resource, string, *envoy_admin_v3.Clusters)
|
|
|
|
// MakeValidate is a builtinextensiontemplate.PluginConstructor for a builtinextensiontemplate.EnvoyExtension.
|
|
func MakeValidate(ext extensioncommon.RuntimeConfig) (extensioncommon.BasicExtension, error) {
|
|
var resultErr error
|
|
var plugin Validate
|
|
|
|
if name := ext.EnvoyExtension.Name; name != api.BuiltinValidateExtension {
|
|
return nil, fmt.Errorf("expected extension name '%s' but got %q", api.BuiltinValidateExtension, name)
|
|
}
|
|
|
|
envoyID, _ := ext.EnvoyExtension.Arguments["envoyID"]
|
|
mainEnvoyID, _ := envoyID.(string)
|
|
vip := ""
|
|
snis := map[string]struct{}{}
|
|
upstream, ok := ext.Upstreams[ext.ServiceName]
|
|
if ok {
|
|
vip = upstream.VIP
|
|
if upstream.SNIs == nil || len(upstream.SNIs) == 0 {
|
|
return nil, fmt.Errorf("no SNIs were set, unable to validate Envoy clusters")
|
|
}
|
|
snis = upstream.SNIs
|
|
}
|
|
if mainEnvoyID == "" && vip == "" {
|
|
return nil, fmt.Errorf("envoyID or virtual IP is required")
|
|
}
|
|
plugin.vip = vip
|
|
plugin.envoyID = mainEnvoyID
|
|
plugin.snis = snis
|
|
plugin.resources = make(map[string]*resource)
|
|
|
|
return &plugin, resultErr
|
|
}
|
|
|
|
type Messages []Message
|
|
|
|
type Message struct {
|
|
Success bool
|
|
Message string
|
|
PossibleActions []string
|
|
}
|
|
|
|
func (m Messages) Success() bool {
|
|
for _, message := range m {
|
|
if !message.Success {
|
|
return false
|
|
}
|
|
}
|
|
|
|
return true
|
|
}
|
|
|
|
func (m Messages) Errors() Messages {
|
|
var errors Messages
|
|
for _, message := range m {
|
|
if !message.Success {
|
|
errors = append(errors, message)
|
|
}
|
|
}
|
|
|
|
return errors
|
|
}
|
|
|
|
// GetMessages returns the error based only on Validate's state.
|
|
func (v *Validate) GetMessages(validateEndpoints bool, endpointValidator EndpointValidator, clusters *envoy_admin_v3.Clusters) Messages {
|
|
var messages Messages
|
|
missingXDSActions := []string{
|
|
"Check that your upstream service is registered with Consul",
|
|
"Make sure your upstream exists by running the `consul[-k8s] troubleshoot upstreams` command",
|
|
"If you are using transparent proxy for this upstream, ensure you have set up allow intentions to the upstream",
|
|
"Check the logs of the Consul agent configuring the local proxy to ensure XDS resources were sent by Consul",
|
|
}
|
|
missingEndpointsActions := []string{
|
|
"Check that your upstream service is healthy and running",
|
|
"Check that your upstream service is registered with Consul",
|
|
"Check that the upstream proxy is healthy and running",
|
|
"If you are explicitly configuring upstreams, ensure the name of the upstream is correct",
|
|
}
|
|
|
|
var upstream string
|
|
upstream = v.envoyID
|
|
if v.envoyID == "" {
|
|
upstream = v.vip
|
|
}
|
|
|
|
if !v.listener {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("No listener for upstream %q", upstream),
|
|
PossibleActions: missingXDSActions,
|
|
})
|
|
} else {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("Listener for upstream %q found", upstream),
|
|
Success: true,
|
|
})
|
|
}
|
|
|
|
if v.usesRDS && !v.route {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("No route for upstream %q", upstream),
|
|
PossibleActions: missingXDSActions,
|
|
})
|
|
} else if v.route {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("Route for upstream %q found", upstream),
|
|
Success: true,
|
|
})
|
|
}
|
|
|
|
numRequiredResources := 0
|
|
// Resources will be marked as required in PatchFilter or PatchRoute because the listener or route will determine
|
|
// which clusters/endpoints to validate.
|
|
for sni, resource := range v.resources {
|
|
if !resource.required {
|
|
continue
|
|
}
|
|
numRequiredResources += 1
|
|
|
|
_, ok := v.snis[sni]
|
|
if !ok || !resource.cluster {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("No cluster %q for upstream %q", sni, upstream),
|
|
PossibleActions: missingXDSActions,
|
|
})
|
|
continue
|
|
} else {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("Cluster %q for upstream %q found", sni, upstream),
|
|
Success: true,
|
|
})
|
|
}
|
|
|
|
// If the resource is a passthrough cluster, it will not have endpoints, so we need to skip the endpoint
|
|
// validation.
|
|
if strings.Contains(sni, "passthrough~") {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("Cluster %q is a passthrough cluster, skipping endpoint healthiness check", sni),
|
|
Success: true,
|
|
})
|
|
continue
|
|
}
|
|
|
|
if validateEndpoints {
|
|
// If resource is a top-level cluster (any cluster that is an aggregate cluster or not a child of an aggregate
|
|
// cluster), it will have an empty parent. If resource is a child cluster, it will have a nonempty parent.
|
|
if resource.parentCluster == "" && resource.aggregateCluster {
|
|
// Aggregate cluster case: do endpoint verification by checking each child cluster. We need at least one
|
|
// child cluster to have healthy endpoints.
|
|
oneClusterHasEndpoints := false
|
|
for _, childCluster := range resource.aggregateClusterChildren {
|
|
endpointValidator(v.resources[childCluster], childCluster, clusters)
|
|
if v.resources[childCluster].endpoints > 0 {
|
|
oneClusterHasEndpoints = true
|
|
}
|
|
}
|
|
if !oneClusterHasEndpoints {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("No healthy endpoints for aggregate cluster %q for upstream %q", sni, upstream),
|
|
PossibleActions: missingEndpointsActions,
|
|
})
|
|
} else {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("Healthy endpoints for aggregate cluster %q for upstream %q found", sni, upstream),
|
|
Success: true,
|
|
})
|
|
}
|
|
} else if resource.parentCluster == "" {
|
|
// Top-level non-aggregate cluster case: check for load assignment and healthy endpoints.
|
|
endpointValidator(resource, sni, clusters)
|
|
if (resource.usesEDS && !resource.loadAssignment) || resource.endpoints == 0 {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("No healthy endpoints for cluster %q for upstream %q", sni, upstream),
|
|
PossibleActions: missingEndpointsActions,
|
|
})
|
|
} else {
|
|
messages = append(messages, Message{
|
|
Message: fmt.Sprintf("Healthy endpoints for cluster %q for upstream %q found", sni, upstream),
|
|
Success: true,
|
|
})
|
|
}
|
|
} else {
|
|
// Child cluster case: skip, since it'll be verified by the parent aggregate cluster.
|
|
continue
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
if numRequiredResources == 0 {
|
|
messages = append(messages, Message{Message: fmt.Sprintf("No clusters found on route or listener")})
|
|
}
|
|
|
|
return messages
|
|
}
|
|
|
|
// DoEndpointValidation implements the EndpointVerifier function type.
|
|
func DoEndpointValidation(r *resource, sni string, clusters *envoy_admin_v3.Clusters) {
|
|
clusterStatuses := clusters.GetClusterStatuses()
|
|
if clusterStatuses == nil {
|
|
return
|
|
}
|
|
status := &envoy_admin_v3.ClusterStatus{}
|
|
r.loadAssignment = false
|
|
for _, s := range clusterStatuses {
|
|
if s.Name == sni {
|
|
status = s
|
|
r.loadAssignment = true
|
|
break
|
|
}
|
|
}
|
|
|
|
healthyEndpoints := 0
|
|
hostStatuses := status.GetHostStatuses()
|
|
|
|
if r.loadAssignment && hostStatuses != nil {
|
|
for _, h := range hostStatuses {
|
|
health := h.GetHealthStatus()
|
|
if health != nil {
|
|
if health.EdsHealthStatus == envoy_core_v3.HealthStatus_HEALTHY && health.FailedOutlierCheck == false {
|
|
healthyEndpoints += 1
|
|
}
|
|
}
|
|
}
|
|
}
|
|
r.endpoints = healthyEndpoints
|
|
}
|
|
|
|
// CanApply determines if the extension can apply to the given extension configuration.
|
|
func (p *Validate) CanApply(config *extensioncommon.RuntimeConfig) bool {
|
|
return true
|
|
}
|
|
|
|
func (v *Validate) PatchRoute(p extensioncommon.RoutePayload) (*envoy_route_v3.RouteConfiguration, bool, error) {
|
|
route := p.Message
|
|
// Route name on connect proxies will be the envoy ID. We are only validating routes for the specific upstream with
|
|
// the envoyID configured.
|
|
if route.Name != p.Upstream.EnvoyID {
|
|
return route, false, nil
|
|
}
|
|
v.route = true
|
|
for sni := range extensioncommon.RouteClusterNames(route) {
|
|
if _, ok := v.resources[sni]; ok {
|
|
continue
|
|
}
|
|
v.resources[sni] = &resource{required: true}
|
|
}
|
|
return route, false, nil
|
|
}
|
|
|
|
func (v *Validate) PatchCluster(p extensioncommon.ClusterPayload) (*envoy_cluster_v3.Cluster, bool, error) {
|
|
c := p.Message
|
|
val, ok := v.resources[c.Name]
|
|
if !ok {
|
|
val = &resource{}
|
|
v.resources[c.Name] = val
|
|
}
|
|
val.cluster = true
|
|
|
|
// If it's an aggregate cluster, add the child clusters to p.resources if they are not already there.
|
|
aggregateCluster, ok := isAggregateCluster(c)
|
|
if ok {
|
|
// Mark this as an aggregate cluster, so we know we do not need to validate its endpoints directly.
|
|
val.aggregateCluster = true
|
|
for _, clusterName := range aggregateCluster.Clusters {
|
|
r, ok := v.resources[clusterName]
|
|
if !ok {
|
|
r = &resource{}
|
|
v.resources[clusterName] = r
|
|
}
|
|
if val.aggregateClusterChildren == nil {
|
|
val.aggregateClusterChildren = []string{}
|
|
}
|
|
// On the parent cluster, add the children.
|
|
val.aggregateClusterChildren = append(val.aggregateClusterChildren, clusterName)
|
|
// On the child cluster, set the parent.
|
|
r.parentCluster = c.Name
|
|
// The child clusters of an aggregate cluster will be required if the parent cluster is.
|
|
r.required = val.required
|
|
}
|
|
return c, false, nil
|
|
}
|
|
|
|
if c.EdsClusterConfig != nil {
|
|
val.usesEDS = true
|
|
} else {
|
|
la := c.LoadAssignment
|
|
if la == nil {
|
|
return c, false, nil
|
|
}
|
|
val.endpoints = len(la.Endpoints) + len(la.NamedEndpoints)
|
|
}
|
|
return c, false, nil
|
|
}
|
|
|
|
func (v *Validate) PatchFilter(p extensioncommon.FilterPayload) (*envoy_listener_v3.Filter, bool, error) {
|
|
filter := p.Message
|
|
// If a single filter exists for a listener we say it exists.
|
|
v.listener = true
|
|
|
|
if httpConfig := envoy_resource_v3.GetHTTPConnectionManager(filter); httpConfig != nil {
|
|
// If the http filter uses RDS, then the clusters we need to validate exist in the route, and there's nothing
|
|
// else we need to do with the filter.
|
|
if httpConfig.GetRds() != nil {
|
|
v.usesRDS = true
|
|
|
|
// Edit the runtime configuration to add an envoy ID based on the route name in the filter. This is because
|
|
// routes are matched by envoyID and in the transparent proxy case, we only have the VIP set in the
|
|
// RuntimeConfig.
|
|
v.envoyID = httpConfig.GetRds().RouteConfigName
|
|
emptyServiceKey := api.CompoundServiceName{}
|
|
upstream, ok := p.RuntimeConfig.Upstreams[emptyServiceKey]
|
|
if ok {
|
|
upstream.EnvoyID = v.envoyID
|
|
}
|
|
return filter, true, nil
|
|
}
|
|
}
|
|
|
|
// FilterClusterNames handles the filter being an http or tcp filter.
|
|
for sni := range extensioncommon.FilterClusterNames(filter) {
|
|
// Mark any clusters we see as required resources.
|
|
if r, ok := v.resources[sni]; ok {
|
|
r.required = true
|
|
} else {
|
|
v.resources[sni] = &resource{required: true}
|
|
}
|
|
}
|
|
|
|
return filter, true, nil
|
|
}
|
|
|
|
func isAggregateCluster(c *envoy_cluster_v3.Cluster) (*envoy_aggregate_cluster_v3.ClusterConfig, bool) {
|
|
aggregateCluster := &envoy_aggregate_cluster_v3.ClusterConfig{}
|
|
cdt, ok := c.ClusterDiscoveryType.(*envoy_cluster_v3.Cluster_ClusterType)
|
|
if ok {
|
|
cct := cdt.ClusterType.TypedConfig
|
|
if cct != nil {
|
|
err := anypb.UnmarshalTo(cct, aggregateCluster, proto.UnmarshalOptions{})
|
|
if err == nil {
|
|
return aggregateCluster, true
|
|
}
|
|
}
|
|
}
|
|
return nil, false
|
|
}
|