Location based Load Balancing (istio#10720)

* support locality weighted loadbalancer

* validate LocalityWeightSettings

* set proxy locality

* fix ci

* address comments

* EDS part

* fix test

* reduce locality string split

* perf optimize

* add ut

* fix lint

* mark TODO

* fix nil pointer

* optimize ApplyLocalityWeightSetting

* per locality edsClusters cache

* update validation ut

* update Locality field

* set locality Lb priority

* fix build

* fix build

* fix panic

* fix minor

* add ut

* address comments

* use cloned endpoints

* fix subset locality lb

* shallow copy cluster to prevent r/w data race

* fix frankbu comments

* update Validate

* update locality lb set

* fix ut

* bump istio api

* update

* fix nil pointer

* fix build

* add comments

* perf improve

* add more comments

* move validate localityLbsetting to validation.go

* fix lint

istioctl/cmd/istioctl/kubeinject.go

@@ -254,6 +254,10 @@ istioctl kube-inject -f deployment.yaml -o deployment-injected.yaml --injectConf
 					return err
 				}
 			}
+			err = model.ValidateMeshConfig(meshConfig)
+			if err != nil {
+				return err
+			}
 
 			var sidecarTemplate string
 

mixer/test/client/gateway/gateway_test.go

@@ -223,6 +223,9 @@ func (mock) ID(*core.Node) string {
 func (mock) GetProxyServiceInstances(_ *model.Proxy) ([]*model.ServiceInstance, error) {
 	return nil, nil
 }
+func (mock) GetProxyLocality(_ *model.Proxy) string {
+	return ""
+}
 func (mock) GetService(_ model.Hostname) (*model.Service, error) { return nil, nil }
 func (mock) InstancesByPort(_ model.Hostname, _ int, _ model.LabelsCollection) ([]*model.ServiceInstance, error) {
 	return nil, nil

mixer/test/client/pilotplugin/pilotplugin_test.go

@@ -297,6 +297,9 @@ func (mock) ID(*core.Node) string {
 func (mock) GetProxyServiceInstances(_ *model.Proxy) ([]*model.ServiceInstance, error) {
 	return nil, nil
 }
+func (mock) GetProxyLocality(_ *model.Proxy) string {
+	return ""
+}
 func (mock) GetService(_ model.Hostname) (*model.Service, error) { return nil, nil }
 func (mock) InstancesByPort(_ model.Hostname, _ int, _ model.LabelsCollection) ([]*model.ServiceInstance, error) {
 	return nil, nil

mixer/test/client/pilotplugin_mtls/pilotplugin_mtls_test.go

@@ -312,6 +312,9 @@ func (mock) ID(*core.Node) string {
 func (mock) GetProxyServiceInstances(_ *model.Proxy) ([]*model.ServiceInstance, error) {
 	return nil, nil
 }
+func (mock) GetProxyLocality(_ *model.Proxy) string {
+	return ""
+}
 func (mock) GetService(_ model.Hostname) (*model.Service, error) { return nil, nil }
 func (mock) InstancesByPort(_ model.Hostname, _ int, _ model.LabelsCollection) ([]*model.ServiceInstance, error) {
 	return nil, nil

mixer/test/client/pilotplugin_tcp/pilotplugin_tcp_test.go

@@ -174,6 +174,9 @@ func (mock) ID(*core.Node) string {
 func (mock) GetProxyServiceInstances(_ *model.Proxy) ([]*model.ServiceInstance, error) {
 	return nil, nil
 }
+func (mock) GetProxyLocality(_ *model.Proxy) string {
+	return ""
+}
 func (mock) GetService(_ model.Hostname) (*model.Service, error) { return nil, nil }
 func (mock) InstancesByPort(_ model.Hostname, _ int, _ model.LabelsCollection) ([]*model.ServiceInstance, error) {
 	return nil, nil

pilot/pkg/bootstrap/server.go

@@ -410,6 +410,11 @@ func (s *Server) initMesh(args *PilotArgs) error {
 		}
 	}
 
+	if err = model.ValidateMeshConfig(mesh); err != nil {
+		log.Errorf("invalid mesh configuration: %v", err)
+		return err
+	}
+
 	log.Infof("mesh configuration %s", spew.Sdump(mesh))
 	log.Infof("version %s", version.Info.String())
 	log.Infof("flags %s", spew.Sdump(args))

pilot/pkg/kube/inject/webhook.go

@@ -36,6 +36,7 @@ import (
 
 	meshconfig "istio.io/api/mesh/v1alpha1"
 	"istio.io/istio/pilot/cmd"
+	"istio.io/istio/pilot/pkg/model"
 	"istio.io/istio/pkg/log"
 )
 
@@ -87,6 +88,10 @@ func loadConfig(injectFile, meshFile string) (*Config, *meshconfig.MeshConfig, e
 	if err != nil {
 		return nil, nil, err
 	}
+	err = model.ValidateMeshConfig(meshConfig)
+	if err != nil {
+		return nil, nil, err
+	}
 
 	log.Infof("New configuration: sha256sum %x", sha256.Sum256(data))
 	log.Infof("Policy: %v", c.Policy)

pilot/pkg/model/context.go

@@ -21,6 +21,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/envoyproxy/go-control-plane/envoy/api/v2/core"
 	"github.com/gogo/protobuf/types"
 	multierror "github.com/hashicorp/go-multierror"
 
@@ -86,6 +87,9 @@ type Proxy struct {
 	// namespace.
 	ID string
 
+	// Locality is the location of where Envoy proxy runs.
+	Locality Locality
+
 	// DNSDomain defines the DNS domain suffix for short hostnames (e.g.
 	// "default.svc.cluster.local")
 	DNSDomain string
@@ -282,6 +286,19 @@ func GetProxyConfigNamespace(proxy *Proxy) string {
 	return ""
 }
 
+// GetProxyLocality returns the locality where Envoy proxy is running.
+func GetProxyLocality(proxy *core.Node) *Locality {
+	if proxy == nil || proxy.Locality == nil {
+		return nil
+	}
+
+	return &Locality{
+		Region:  proxy.Locality.Region,
+		Zone:    proxy.Locality.Zone,
+		SubZone: proxy.Locality.SubZone,
+	}
+}
+
 const (
 	serviceNodeSeparator = "~"
 

pilot/pkg/model/locality.go

@@ -0,0 +1,61 @@
+// Copyright 2019 Istio 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 model
+
+type LocalityInterface interface {
+	GetRegion() string
+	GetZone() string
+	GetSubZone() string
+}
+
+// Identifies location of where either Envoy runs or where upstream hosts run.
+type Locality struct {
+	// Region this proxy belongs to.
+	Region string
+	// Defines the local service zone where Envoy is running. Though optional, it
+	// should be set if discovery service routing is used and the discovery
+	// service exposes :ref:`zone data <envoy_api_field_endpoint.LocalityLbEndpoints.locality>`,
+	// either in this message or via :option:`--service-zone`. The meaning of zone
+	// is context dependent, e.g. `Availability Zone (AZ)
+	// <https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-regions-availability-zones.html>`_
+	// on AWS, `Zone <https://cloud.google.com/compute/docs/regions-zones/>`_ on
+	// GCP, etc.
+	Zone string
+	// When used for locality of upstream hosts, this field further splits zone
+	// into smaller chunks of sub-zones so they can be load balanced
+	// independently.
+	SubZone string
+}
+
+func (l *Locality) GetRegion() string {
+	if l != nil {
+		return l.Region
+	}
+	return ""
+}
+
+func (l *Locality) GetZone() string {
+	if l != nil {
+		return l.Zone
+	}
+	return ""
+}
+
+func (l *Locality) GetSubZone() string {
+	if l != nil {
+		return l.SubZone
+	}
+	return ""
+}

pilot/pkg/model/push_context.go

@@ -480,6 +480,17 @@ func (ps *PushContext) GetAllSidecarScopes() map[string][]*SidecarScope {
 	return ps.sidecarsByNamespace
 }
 
+// ServicePort returns the port model for the given service and port.
+func (ps *PushContext) ServicePort(hostname Hostname, port int) *Port {
+	portList := ps.ServicePort2Name[string(hostname)]
+	for i := range portList {
+		if portList[i] != nil && portList[i].Port == port {
+			return portList[i]
+		}
+	}
+	return nil
+}
+
 // DestinationRule returns a destination rule for a service name in a given domain.
 func (ps *PushContext) DestinationRule(proxy *Proxy, hostname Hostname) *Config {
 	// If proxy has a sidecar scope that is user supplied, then get the destination rules from the sidecar scope

pilot/pkg/model/service.go

@@ -496,6 +496,9 @@ type ServiceDiscovery interface {
 	// determine the intended destination of a connection without a Host header on the request.
 	GetProxyServiceInstances(*Proxy) ([]*ServiceInstance, error)
 
+	// GetProxyLocality returns the locality where the proxy runs.
+	GetProxyLocality(*Proxy) string
+
 	// ManagementPorts lists set of management ports associated with an IPv4 address.
 	// These management ports are typically used by the platform for out of band management
 	// tasks such as health checks, etc. In a scenario where the proxy functions in the
@@ -535,12 +538,12 @@ func (h Hostname) Matches(o Hostname) bool {
 		return true
 	}
 
-	hWildcard := string(h[0]) == "*"
+	hWildcard := len(h) > 0 && string(h[0]) == "*"
 	if hWildcard && len(o) == 0 {
 		return true
 	}
 
-	oWildcard := string(o[0]) == "*"
+	oWildcard := len(o) > 0 && string(o[0]) == "*"
 	if !hWildcard && !oWildcard {
 		// both are non-wildcards, so do normal string comparison
 		return h == o
@@ -574,8 +577,8 @@ func (h Hostname) SubsetOf(o Hostname) bool {
 		return true
 	}
 
-	hWildcard := string(h[0]) == "*"
-	oWildcard := string(o[0]) == "*"
+	hWildcard := len(h) > 0 && string(h[0]) == "*"
+	oWildcard := len(o) > 0 && string(o[0]) == "*"
 	if !oWildcard {
 		if hWildcard {
 			return false

pilot/pkg/model/service_test.go

@@ -224,6 +224,8 @@ func TestHostnameMatches(t *testing.T) {
 		out  bool
 	}{
 		{"empty", "", "", true},
+		{"first empty", "", "foo.com", false},
+		{"second empty", "foo.com", "", false},
 
 		{"non-wildcard domain",
 			"foo.com", "foo.com", true},

pilot/pkg/model/validation.go

@@ -824,7 +824,6 @@ func validateLoadBalancer(settings *networking.LoadBalancerSettings) (errs error
 			}
 		}
 	}
-
 	return
 }
 
@@ -1044,6 +1043,10 @@ func ValidateMeshConfig(mesh *meshconfig.MeshConfig) (errs error) {
 		errs = multierror.Append(errs, err)
 	}
 
+	if err := validateLocalityLbSetting(mesh.LocalityLbSetting); err != nil {
+		errs = multierror.Append(errs, err)
+	}
+
 	return
 }
 
@@ -2214,3 +2217,111 @@ func ValidateNetworkEndpointAddress(n *NetworkEndpoint) error {
 	}
 	return nil
 }
+
+// validateLocalityLbSetting checks the LocalityLbSetting of MeshConfig
+func validateLocalityLbSetting(lb *meshconfig.LocalityLoadBalancerSetting) error {
+	if lb == nil {
+		return nil
+	}
+
+	if len(lb.GetDistribute()) > 0 && len(lb.GetFailover()) > 0 {
+		return fmt.Errorf("can not simultaneously specify 'distribute' and 'failover'")
+	}
+
+	srcLocalities := []string{}
+	for _, locality := range lb.GetDistribute() {
+		srcLocalities = append(srcLocalities, locality.From)
+		var totalWeight uint32
+		destLocalities := []string{}
+		for loc, weight := range locality.To {
+			destLocalities = append(destLocalities, loc)
+			if weight == 0 {
+				return fmt.Errorf("locality weight must not be in range [1, 100]")
+			}
+			totalWeight += weight
+		}
+		if totalWeight != 100 {
+			return fmt.Errorf("total locality weight %v != 100", totalWeight)
+		}
+		if err := validateLocalities(destLocalities); err != nil {
+			return err
+		}
+	}
+
+	if err := validateLocalities(srcLocalities); err != nil {
+		return err
+	}
+
+	for _, failover := range lb.GetFailover() {
+		if failover.From == failover.To {
+			return fmt.Errorf("locality lb failover settings must specify different regions")
+		}
+		if strings.Contains(failover.To, "*") {
+			return fmt.Errorf("locality lb failover region should not contain '*' wildcard")
+		}
+	}
+
+	return nil
+}
+
+func validateLocalities(localities []string) error {
+	regionZoneSubZoneMap := map[string]map[string]map[string]bool{}
+
+	for _, locality := range localities {
+		if n := strings.Count(locality, "*"); n > 0 {
+			if n > 1 || !strings.HasSuffix(locality, "*") {
+				return fmt.Errorf("locality %s wildcard '*' number can not exceed 1 and must be in the end", locality)
+			}
+		}
+
+		items := strings.SplitN(locality, "/", 3)
+		for _, item := range items {
+			if item == "" {
+				return fmt.Errorf("locality %s must not contain empty region/zone/subzone info", locality)
+			}
+		}
+		if _, ok := regionZoneSubZoneMap["*"]; ok {
+			return fmt.Errorf("locality %s overlap with previous specified ones", locality)
+		}
+		switch len(items) {
+		case 1:
+			if _, ok := regionZoneSubZoneMap[items[0]]; ok {
+				return fmt.Errorf("locality %s overlap with previous specified ones", locality)
+			}
+			regionZoneSubZoneMap[items[0]] = map[string]map[string]bool{"*": {"*": true}}
+		case 2:
+			if _, ok := regionZoneSubZoneMap[items[0]]; ok {
+				if _, ok := regionZoneSubZoneMap[items[0]]["*"]; ok {
+					return fmt.Errorf("locality %s overlap with previous specified ones", locality)
+				}
+				if _, ok := regionZoneSubZoneMap[items[0]][items[1]]; ok {
+					return fmt.Errorf("locality %s overlap with previous specified ones", locality)
+				}
+				regionZoneSubZoneMap[items[0]][items[1]] = map[string]bool{"*": true}
+			} else {
+				regionZoneSubZoneMap[items[0]] = map[string]map[string]bool{items[1]: {"*": true}}
+			}
+		case 3:
+			if _, ok := regionZoneSubZoneMap[items[0]]; ok {
+				if _, ok := regionZoneSubZoneMap[items[0]]["*"]; ok {
+					return fmt.Errorf("locality %s overlap with previous specified ones", locality)
+				}
+				if _, ok := regionZoneSubZoneMap[items[0]][items[1]]; ok {
+					if regionZoneSubZoneMap[items[0]][items[1]]["*"] {
+						return fmt.Errorf("locality %s overlap with previous specified ones", locality)
+					}
+					if regionZoneSubZoneMap[items[0]][items[1]][items[2]] {
+						return fmt.Errorf("locality %s overlap with previous specified ones", locality)
+					}
+					regionZoneSubZoneMap[items[0]][items[1]][items[2]] = true
+				} else {
+					regionZoneSubZoneMap[items[0]][items[1]] = map[string]bool{items[2]: true}
+				}
+			} else {
+				regionZoneSubZoneMap[items[0]] = map[string]map[string]bool{items[1]: {items[2]: true}}
+			}
+		}
+	}
+
+	return nil
+}