Merge pull request #83 from numberly/feat/lb-weight

Feature : Add a new annotation to configure envoy upstream weight

Author: mmcgarr

.gitignore

@@ -1,3 +1,7 @@
 bin/
 command
 testing
+ca/
+config/
+envoy/
+

README.md

@@ -76,6 +76,7 @@ Yggdrasil allows for some customisation of the route and cluster config per Ingr
 |--------------------------------------------------------------|----------|
 | [yggdrasil.uswitch.com/healthcheck-path](#health-check-path) | string   |
 | [yggdrasil.uswitch.com/timeout](#timeout)                    | duration |
+| [yggdrasil.uswitch.com/weight](#weight)                      | uint32   |
 | [yggdrasil.uswitch.com/retry-on](#retries)                   | string   |
 
 ### Health Check Path
@@ -90,6 +91,9 @@ Allows for adjusting the timeout in envoy. Currently this will set the following
 * [config.route.v3.RetryPolicy.PerTryTimeout](https://www.envoyproxy.io/docs/envoy/v1.19.0/api-v3/config/route/v3/route_components.proto#envoy-v3-api-field-config-route-v3-retrypolicy-per-try-timeout)
 * [config.cluster.v3.Cluster.ConnectTimeout](https://www.envoyproxy.io/docs/envoy/v1.19.0/api-v3/config/cluster/v3/cluster.proto#envoy-v3-api-field-config-cluster-v3-cluster-connect-timeout)
 
+### Weight
+Allows for adjusting the [load balancer weights](https://www.envoyproxy.io/docs/envoy/latest/api-v3/config/endpoint/v3/endpoint_components.proto#config-endpoint-v3-lbendpoint) in envoy.
+
 ### Retries
 Allows overwriting the default retry policy's [config.route.v3.RetryPolicy.RetryOn](https://www.envoyproxy.io/docs/envoy/v1.19.0/api-v3/config/route/v3/route_components.proto#envoy-v3-api-field-config-route-v3-retrypolicy-retry-on) set by the `--retry-on` flag (default 5xx). Accepts a comma-separated list of retry-on policies.
 
@@ -105,6 +109,7 @@ metadata:
   annotations:
     yggdrasil.uswitch.com/healthcheck-path: /healthz
     yggdrasil.uswitch.com/timeout: 30s
+    yggdrasil.uswitch.com/weight: "12"
     yggdrasil.uswitch.com/retry-on: gateway-error,connect-failure
 spec:
   rules:

pkg/envoy/boilerplate.go

@@ -400,14 +400,14 @@ func makeListener(filterChains []*listener.FilterChain, envoyListenerIpv4Address
 	return &listener, nil
 }
 
-func makeAddresses(addresses []string, upstreamPort uint32) []*core.Address {
+func makeAddresses(addresses []LBHost, upstreamPort uint32) []*core.Address {
 
 	envoyAddresses := []*core.Address{}
 	for _, address := range addresses {
 		envoyAddress := &core.Address{
 			Address: &core.Address_SocketAddress{
 				SocketAddress: &core.SocketAddress{
-					Address: address,
+					Address: address.Host,
 					PortSpecifier: &core.SocketAddress_PortValue{
 						PortValue: upstreamPort,
 					},
@@ -475,7 +475,8 @@ func makeCluster(c cluster, ca string, healthCfg UpstreamHealthCheck, outlierPer
 
 	for idx, address := range addresses {
 		endpoints[idx] = &endpoint.LbEndpoint{
-			HostIdentifier: &endpoint.LbEndpoint_Endpoint{Endpoint: &endpoint.Endpoint{Address: address}},
+			HostIdentifier:      &endpoint.LbEndpoint_Endpoint{Endpoint: &endpoint.Endpoint{Address: address}},
+			LoadBalancingWeight: &wrappers.UInt32Value{Value: c.Hosts[idx].Weight},
 		}
 	}
 

pkg/envoy/ingress_translator.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"regexp"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -90,12 +91,17 @@ func (v *virtualHost) Equals(other *virtualHost) bool {
 		v.RetryOn == other.RetryOn
 }
 
+type LBHost struct {
+	Host   string
+	Weight uint32
+}
+
 type cluster struct {
 	Name            string
 	VirtualHost     string
 	HealthCheckPath string
 	Timeout         time.Duration
-	Hosts           []string
+	Hosts           []LBHost
 }
 
 func (c *cluster) identity() string {
@@ -127,8 +133,12 @@ func (c *cluster) Equals(other *cluster) bool {
 		return false
 	}
 
-	sort.Strings(c.Hosts)
-	sort.Strings(other.Hosts)
+	sort.Slice(c.Hosts[:], func(i, j int) bool {
+		return c.Hosts[i].Host < c.Hosts[j].Host
+	})
+	sort.Slice(other.Hosts[:], func(i, j int) bool {
+		return other.Hosts[i].Host < other.Hosts[j].Host
+	})
 
 	for i, host := range c.Hosts {
 		if host != other.Hosts[i] {
@@ -197,15 +207,15 @@ func newEnvoyIngress(host string) *envoyIngress {
 		cluster: &cluster{
 			Name:            clusterName,
 			VirtualHost:     host,
-			Hosts:           []string{},
+			Hosts:           []LBHost{},
 			Timeout:         (30 * time.Second),
 			HealthCheckPath: "",
 		},
 	}
 }
 
-func (ing *envoyIngress) addUpstream(host string) {
-	ing.cluster.Hosts = append(ing.cluster.Hosts, host)
+func (ing *envoyIngress) addUpstream(host string, weight uint32) {
+	ing.cluster.Hosts = append(ing.cluster.Hosts, LBHost{host, weight})
 }
 
 func (ing *envoyIngress) addHealthCheckPath(path string) {
@@ -311,7 +321,14 @@ func translateIngresses(ingresses []*k8s.Ingress, syncSecrets bool, secrets []*v
 				}
 
 				envoyIngress := envoyIngresses[ruleHost]
-				envoyIngress.addUpstream(j)
+
+				if weight64, err := strconv.ParseUint(i.Annotations["yggdrasil.uswitch.com/weight"], 10, 32); err == nil {
+					if weight64 != 0 {
+						envoyIngress.addUpstream(j, uint32(weight64))
+					}
+				} else {
+					envoyIngress.addUpstream(j, 1)
+				}
 
 				if i.Annotations["yggdrasil.uswitch.com/healthcheck-path"] != "" {
 					envoyIngress.addHealthCheckPath(i.Annotations["yggdrasil.uswitch.com/healthcheck-path"])

pkg/envoy/ingress_translator_test.go

@@ -131,8 +131,8 @@ func TestVirtualHostEquality(t *testing.T) {
 }
 
 func TestClusterEquality(t *testing.T) {
-	a := &cluster{Name: "foo", Hosts: []string{"host1", "host2"}}
-	b := &cluster{Name: "foo", Hosts: []string{"host1", "host2"}}
+	a := &cluster{Name: "foo", Hosts: []LBHost{{"host1", 1}, {"host2", 1}}}
+	b := &cluster{Name: "foo", Hosts: []LBHost{{"host1", 1}, {"host2", 1}}}
 
 	if !a.Equals(b) {
 		t.Error()
@@ -142,17 +142,17 @@ func TestClusterEquality(t *testing.T) {
 		t.Error("cluster is equals nil, expect not to be equal")
 	}
 
-	c := &cluster{Name: "bar", Hosts: []string{"host1", "host2"}}
+	c := &cluster{Name: "bar", Hosts: []LBHost{{"host1", 1}, {"host2", 1}}}
 	if a.Equals(c) {
 		t.Error("clusters have different names, expected not to be equal")
 	}
 
-	d := &cluster{Name: "foo", Hosts: []string{"host1"}} // missing host2
+	d := &cluster{Name: "foo", Hosts: []LBHost{{"host1", 1}}} // missing host2
 	if a.Equals(d) {
 		t.Error("clusters have different hosts, should be different")
 	}
 
-	e := &cluster{Name: "foo", Hosts: []string{"bad1", "bad2"}}
+	e := &cluster{Name: "foo", Hosts: []LBHost{{"bad1", 1}, {"bad2", 1}}}
 	if a.Equals(e) {
 		t.Error("cluster hosts are different, shouldn't be equal")
 	}
@@ -162,7 +162,7 @@ func TestClusterEquality(t *testing.T) {
 		t.Error("no hosts set")
 	}
 
-	g := &cluster{Name: "foo", Hosts: []string{"host1", "host2"}, Timeout: (5 * time.Second)}
+	g := &cluster{Name: "foo", Hosts: []LBHost{{"host1", 1}, {"host2", 1}}, Timeout: (5 * time.Second)}
 	if a.Equals(g) {
 		t.Error("clusters with different timeout values should not be equal")
 	}
@@ -268,8 +268,12 @@ func TestGeneratesForSingleIngress(t *testing.T) {
 	if c.Clusters[0].Name != "foo_app_com" {
 		t.Errorf("expected cluster to be named after ingress host, was %s", c.Clusters[0].Name)
 	}
-	if c.Clusters[0].Hosts[0] != "foo.cluster.com" {
-		t.Errorf("expected cluster host for foo.cluster.com, was %s", c.Clusters[0].Hosts[0])
+	if c.Clusters[0].Hosts[0].Host != "foo.cluster.com" {
+		t.Errorf("expected cluster host for foo.cluster.com, was %s", c.Clusters[0].Hosts[0].Host)
+	}
+
+	if c.Clusters[0].Hosts[0].Weight != 1 {
+		t.Errorf("expected cluster host's weight for 1, was %d", c.Clusters[0].Hosts[0].Weight)
 	}
 
 	if c.VirtualHosts[0].UpstreamCluster != c.Clusters[0].Name {
@@ -304,11 +308,11 @@ func TestGeneratesForMultipleIngressSharingSpecHost(t *testing.T) {
 	if len(c.Clusters[0].Hosts) != 2 {
 		t.Errorf("expected 2 host, was %d", len(c.Clusters[0].Hosts))
 	}
-	if c.Clusters[0].Hosts[0] != "foo.com" {
-		t.Errorf("expected cluster host for foo.com, was %s", c.Clusters[0].Hosts[0])
+	if c.Clusters[0].Hosts[0].Host != "foo.com" {
+		t.Errorf("expected cluster host for foo.com, was %s", c.Clusters[0].Hosts[0].Host)
 	}
-	if c.Clusters[0].Hosts[1] != "bar.com" {
-		t.Errorf("expected cluster host for bar.com, was %s", c.Clusters[0].Hosts[1])
+	if c.Clusters[0].Hosts[1].Host != "bar.com" {
+		t.Errorf("expected cluster host for bar.com, was %s", c.Clusters[0].Hosts[1].Host)
 	}
 
 	if c.VirtualHosts[0].UpstreamCluster != c.Clusters[0].Name {
@@ -340,8 +344,8 @@ func TestFilterNonMatchingIngresses(t *testing.T) {
 func TestIngressWithIP(t *testing.T) {
 	ingress := newIngressIP("app.com", "127.0.0.1")
 	c := translateIngresses([]*k8s.Ingress{ingress}, false, []*v1.Secret{})
-	if c.Clusters[0].Hosts[0] != "127.0.0.1" {
-		t.Errorf("expected cluster host to be IP address, was %s", c.Clusters[0].Hosts[0])
+	if c.Clusters[0].Hosts[0].Host != "127.0.0.1" {
+		t.Errorf("expected cluster host to be IP address, was %s", c.Clusters[0].Hosts[0].Host)
 	}
 }