unmarshal_rds.go

/*
 *
 * Copyright 2021 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package xdsresource

import (
	"fmt"
	"regexp"
	"strings"
	"time"

	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
	v3typepb "github.com/envoyproxy/go-control-plane/envoy/type/v3"
	"github.com/golang/protobuf/proto"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/internal/grpclog"
	"google.golang.org/grpc/internal/pretty"
	"google.golang.org/grpc/internal/xds/env"
	"google.golang.org/grpc/xds/internal/version"
	"google.golang.org/protobuf/types/known/anypb"
)

// UnmarshalRouteConfig processes resources received in an RDS response,
// validates them, and transforms them into a native struct which contains only
// fields we are interested in. The provided hostname determines the route
// configuration resources of interest.
func UnmarshalRouteConfig(opts *UnmarshalOptions) (map[string]RouteConfigUpdateErrTuple, UpdateMetadata, error) {
	update := make(map[string]RouteConfigUpdateErrTuple)
	md, err := processAllResources(opts, update)
	return update, md, err
}

func unmarshalRouteConfigResource(r *anypb.Any, logger *grpclog.PrefixLogger) (string, RouteConfigUpdate, error) {
	if !IsRouteConfigResource(r.GetTypeUrl()) {
		return "", RouteConfigUpdate{}, fmt.Errorf("unexpected resource type: %q ", r.GetTypeUrl())
	}
	rc := &v3routepb.RouteConfiguration{}
	if err := proto.Unmarshal(r.GetValue(), rc); err != nil {
		return "", RouteConfigUpdate{}, fmt.Errorf("failed to unmarshal resource: %v", err)
	}
	logger.Infof("Resource with name: %v, type: %T, contains: %v.", rc.GetName(), rc, pretty.ToJSON(rc))

	// TODO: Pass version.TransportAPI instead of relying upon the type URL
	v2 := r.GetTypeUrl() == version.V2RouteConfigURL
	u, err := generateRDSUpdateFromRouteConfiguration(rc, logger, v2)
	if err != nil {
		return rc.GetName(), RouteConfigUpdate{}, err
	}
	u.Raw = r
	return rc.GetName(), u, nil
}

// generateRDSUpdateFromRouteConfiguration checks if the provided
// RouteConfiguration meets the expected criteria. If so, it returns a
// RouteConfigUpdate with nil error.
//
// A RouteConfiguration resource is considered valid when only if it contains a
// VirtualHost whose domain field matches the server name from the URI passed
// to the gRPC channel, and it contains a clusterName or a weighted cluster.
//
// The RouteConfiguration includes a list of virtualHosts, which may have zero
// or more elements. We are interested in the element whose domains field
// matches the server name specified in the "xds:" URI. The only field in the
// VirtualHost proto that the we are interested in is the list of routes. We
// only look at the last route in the list (the default route), whose match
// field must be empty and whose route field must be set.  Inside that route
// message, the cluster field will contain the clusterName or weighted clusters
// we are looking for.
func generateRDSUpdateFromRouteConfiguration(rc *v3routepb.RouteConfiguration, logger *grpclog.PrefixLogger, v2 bool) (RouteConfigUpdate, error) {
	vhs := make([]*VirtualHost, 0, len(rc.GetVirtualHosts()))
	for _, vh := range rc.GetVirtualHosts() {
		routes, err := routesProtoToSlice(vh.Routes, logger, v2)
		if err != nil {
			return RouteConfigUpdate{}, fmt.Errorf("received route is invalid: %v", err)
		}
		rc, err := generateRetryConfig(vh.GetRetryPolicy())
		if err != nil {
			return RouteConfigUpdate{}, fmt.Errorf("received route is invalid: %v", err)
		}
		vhOut := &VirtualHost{
			Domains:     vh.GetDomains(),
			Routes:      routes,
			RetryConfig: rc,
		}
		if !v2 {
			cfgs, err := processHTTPFilterOverrides(vh.GetTypedPerFilterConfig())
			if err != nil {
				return RouteConfigUpdate{}, fmt.Errorf("virtual host %+v: %v", vh, err)
			}
			vhOut.HTTPFilterConfigOverride = cfgs
		}
		vhs = append(vhs, vhOut)
	}
	return RouteConfigUpdate{VirtualHosts: vhs}, nil
}

func generateRetryConfig(rp *v3routepb.RetryPolicy) (*RetryConfig, error) {
	if rp == nil {
		return nil, nil
	}

	cfg := &RetryConfig{RetryOn: make(map[codes.Code]bool)}
	for _, s := range strings.Split(rp.GetRetryOn(), ",") {
		switch strings.TrimSpace(strings.ToLower(s)) {
		case "cancelled":
			cfg.RetryOn[codes.Canceled] = true
		case "deadline-exceeded":
			cfg.RetryOn[codes.DeadlineExceeded] = true
		case "internal":
			cfg.RetryOn[codes.Internal] = true
		case "resource-exhausted":
			cfg.RetryOn[codes.ResourceExhausted] = true
		case "unavailable":
			cfg.RetryOn[codes.Unavailable] = true
		}
	}

	if rp.NumRetries == nil {
		cfg.NumRetries = 1
	} else {
		cfg.NumRetries = rp.GetNumRetries().Value
		if cfg.NumRetries < 1 {
			return nil, fmt.Errorf("retry_policy.num_retries = %v; must be >= 1", cfg.NumRetries)
		}
	}

	backoff := rp.GetRetryBackOff()
	if backoff == nil {
		cfg.RetryBackoff.BaseInterval = 25 * time.Millisecond
	} else {
		cfg.RetryBackoff.BaseInterval = backoff.GetBaseInterval().AsDuration()
		if cfg.RetryBackoff.BaseInterval <= 0 {
			return nil, fmt.Errorf("retry_policy.base_interval = %v; must be > 0", cfg.RetryBackoff.BaseInterval)
		}
	}
	if max := backoff.GetMaxInterval(); max == nil {
		cfg.RetryBackoff.MaxInterval = 10 * cfg.RetryBackoff.BaseInterval
	} else {
		cfg.RetryBackoff.MaxInterval = max.AsDuration()
		if cfg.RetryBackoff.MaxInterval <= 0 {
			return nil, fmt.Errorf("retry_policy.max_interval = %v; must be > 0", cfg.RetryBackoff.MaxInterval)
		}
	}

	if len(cfg.RetryOn) == 0 {
		return &RetryConfig{}, nil
	}
	return cfg, nil
}

func routesProtoToSlice(routes []*v3routepb.Route, logger *grpclog.PrefixLogger, v2 bool) ([]*Route, error) {
	var routesRet []*Route
	for _, r := range routes {
		match := r.GetMatch()
		if match == nil {
			return nil, fmt.Errorf("route %+v doesn't have a match", r)
		}

		if len(match.GetQueryParameters()) != 0 {
			// Ignore route with query parameters.
			logger.Warningf("route %+v has query parameter matchers, the route will be ignored", r)
			continue
		}

		pathSp := match.GetPathSpecifier()
		if pathSp == nil {
			return nil, fmt.Errorf("route %+v doesn't have a path specifier", r)
		}

		var route Route
		switch pt := pathSp.(type) {
		case *v3routepb.RouteMatch_Prefix:
			route.Prefix = &pt.Prefix
		case *v3routepb.RouteMatch_Path:
			route.Path = &pt.Path
		case *v3routepb.RouteMatch_SafeRegex:
			regex := pt.SafeRegex.GetRegex()
			re, err := regexp.Compile(regex)
			if err != nil {
				return nil, fmt.Errorf("route %+v contains an invalid regex %q", r, regex)
			}
			route.Regex = re
		default:
			return nil, fmt.Errorf("route %+v has an unrecognized path specifier: %+v", r, pt)
		}

		if caseSensitive := match.GetCaseSensitive(); caseSensitive != nil {
			route.CaseInsensitive = !caseSensitive.Value
		}

		for _, h := range match.GetHeaders() {
			var header HeaderMatcher
			switch ht := h.GetHeaderMatchSpecifier().(type) {
			case *v3routepb.HeaderMatcher_ExactMatch:
				header.ExactMatch = &ht.ExactMatch
			case *v3routepb.HeaderMatcher_SafeRegexMatch:
				regex := ht.SafeRegexMatch.GetRegex()
				re, err := regexp.Compile(regex)
				if err != nil {
					return nil, fmt.Errorf("route %+v contains an invalid regex %q", r, regex)
				}
				header.RegexMatch = re
			case *v3routepb.HeaderMatcher_RangeMatch:
				header.RangeMatch = &Int64Range{
					Start: ht.RangeMatch.Start,
					End:   ht.RangeMatch.End,
				}
			case *v3routepb.HeaderMatcher_PresentMatch:
				header.PresentMatch = &ht.PresentMatch
			case *v3routepb.HeaderMatcher_PrefixMatch:
				header.PrefixMatch = &ht.PrefixMatch
			case *v3routepb.HeaderMatcher_SuffixMatch:
				header.SuffixMatch = &ht.SuffixMatch
			default:
				return nil, fmt.Errorf("route %+v has an unrecognized header matcher: %+v", r, ht)
			}
			header.Name = h.GetName()
			invert := h.GetInvertMatch()
			header.InvertMatch = &invert
			route.Headers = append(route.Headers, &header)
		}

		if fr := match.GetRuntimeFraction(); fr != nil {
			d := fr.GetDefaultValue()
			n := d.GetNumerator()
			switch d.GetDenominator() {
			case v3typepb.FractionalPercent_HUNDRED:
				n *= 10000
			case v3typepb.FractionalPercent_TEN_THOUSAND:
				n *= 100
			case v3typepb.FractionalPercent_MILLION:
			}
			route.Fraction = &n
		}

		switch r.GetAction().(type) {
		case *v3routepb.Route_Route:
			route.WeightedClusters = make(map[string]WeightedCluster)
			action := r.GetRoute()

			// Hash Policies are only applicable for a Ring Hash LB.
			if env.RingHashSupport {
				hp, err := hashPoliciesProtoToSlice(action.HashPolicy, logger)
				if err != nil {
					return nil, err
				}
				route.HashPolicies = hp
			}

			switch a := action.GetClusterSpecifier().(type) {
			case *v3routepb.RouteAction_Cluster:
				route.WeightedClusters[a.Cluster] = WeightedCluster{Weight: 1}
			case *v3routepb.RouteAction_WeightedClusters:
				wcs := a.WeightedClusters
				var totalWeight uint32
				for _, c := range wcs.Clusters {
					w := c.GetWeight().GetValue()
					if w == 0 {
						continue
					}
					wc := WeightedCluster{Weight: w}
					if !v2 {
						cfgs, err := processHTTPFilterOverrides(c.GetTypedPerFilterConfig())
						if err != nil {
							return nil, fmt.Errorf("route %+v, action %+v: %v", r, a, err)
						}
						wc.HTTPFilterConfigOverride = cfgs
					}
					route.WeightedClusters[c.GetName()] = wc
					totalWeight += w
				}
				// envoy xds doc
				// default TotalWeight https://www.envoyproxy.io/docs/envoy/latest/api-v3/config/route/v3/route_components.proto.html#envoy-v3-api-field-config-route-v3-weightedcluster-total-weight
				wantTotalWeight := uint32(100)
				if tw := wcs.GetTotalWeight(); tw != nil {
					wantTotalWeight = tw.GetValue()
				}
				if totalWeight != wantTotalWeight {
					return nil, fmt.Errorf("route %+v, action %+v, weights of clusters do not add up to total total weight, got: %v, expected total weight from response: %v", r, a, totalWeight, wantTotalWeight)
				}
				if totalWeight == 0 {
					return nil, fmt.Errorf("route %+v, action %+v, has no valid cluster in WeightedCluster action", r, a)
				}
			case *v3routepb.RouteAction_ClusterHeader:
				continue
			default:
				return nil, fmt.Errorf("route %+v, has an unknown ClusterSpecifier: %+v", r, a)
			}

			msd := action.GetMaxStreamDuration()
			// Prefer grpc_timeout_header_max, if set.
			dur := msd.GetGrpcTimeoutHeaderMax()
			if dur == nil {
				dur = msd.GetMaxStreamDuration()
			}
			if dur != nil {
				d := dur.AsDuration()
				route.MaxStreamDuration = &d
			}

			var err error
			route.RetryConfig, err = generateRetryConfig(action.GetRetryPolicy())
			if err != nil {
				return nil, fmt.Errorf("route %+v, action %+v: %v", r, action, err)
			}

			route.RouteAction = RouteActionRoute

		case *v3routepb.Route_NonForwardingAction:
			// Expected to be used on server side.
			route.RouteAction = RouteActionNonForwardingAction
		default:
			route.RouteAction = RouteActionUnsupported
		}

		if !v2 {
			cfgs, err := processHTTPFilterOverrides(r.GetTypedPerFilterConfig())
			if err != nil {
				return nil, fmt.Errorf("route %+v: %v", r, err)
			}
			route.HTTPFilterConfigOverride = cfgs
		}
		routesRet = append(routesRet, &route)
	}
	return routesRet, nil
}

func hashPoliciesProtoToSlice(policies []*v3routepb.RouteAction_HashPolicy, logger *grpclog.PrefixLogger) ([]*HashPolicy, error) {
	var hashPoliciesRet []*HashPolicy
	for _, p := range policies {
		policy := HashPolicy{Terminal: p.Terminal}
		switch p.GetPolicySpecifier().(type) {
		case *v3routepb.RouteAction_HashPolicy_Header_:
			policy.HashPolicyType = HashPolicyTypeHeader
			policy.HeaderName = p.GetHeader().GetHeaderName()
			if rr := p.GetHeader().GetRegexRewrite(); rr != nil {
				regex := rr.GetPattern().GetRegex()
				re, err := regexp.Compile(regex)
				if err != nil {
					return nil, fmt.Errorf("hash policy %+v contains an invalid regex %q", p, regex)
				}
				policy.Regex = re
				policy.RegexSubstitution = rr.GetSubstitution()
			}
		case *v3routepb.RouteAction_HashPolicy_FilterState_:
			if p.GetFilterState().GetKey() != "io.grpc.channel_id" {
				logger.Infof("hash policy %+v contains an invalid key for filter state policy %q", p, p.GetFilterState().GetKey())
				continue
			}
			policy.HashPolicyType = HashPolicyTypeChannelID
		default:
			logger.Infof("hash policy %T is an unsupported hash policy", p.GetPolicySpecifier())
			continue
		}

		hashPoliciesRet = append(hashPoliciesRet, &policy)
	}
	return hashPoliciesRet, nil
}