Merge pull request quarkusio#170 from sgreene570/bz-1858879

Bug 1858879: Change router's internal endpoint.ID to prevent HAProxy server line collisions

pkg/router/metrics/haproxy/haproxy.go

@@ -543,13 +543,17 @@ func knownServerSegment(value string) (string, string, string, bool) {
 	if i := strings.Index(value, ":"); i != -1 {
 		switch value[:i] {
 		case "ept":
-			if service, server, ok := parseNameSegment(value[i+1:]); ok {
-				return "", service, server, true
+			if service, remainder, ok := parseNameSegment(value[i+1:]); ok {
+				if _, server, ok := parseNameSegment(remainder); ok {
+					return "", service, server, true
+				}
 			}
 		case "pod":
 			if pod, remainder, ok := parseNameSegment(value[i+1:]); ok {
-				if service, server, ok := parseNameSegment(remainder); ok {
-					return pod, service, server, true
+				if service, remainder, ok := parseNameSegment(remainder); ok {
+					if _, server, ok := parseNameSegment(remainder); ok {
+						return pod, service, server, true
+					}
 				}
 			}
 		}

pkg/router/template/plugin.go

@@ -322,13 +322,13 @@ func createRouterEndpoints(endpoints *kapi.Endpoints, excludeUDP bool, lookupSvc
 				if a.TargetRef != nil {
 					ep.TargetName = a.TargetRef.Name
 					if a.TargetRef.Kind == "Pod" {
-						ep.ID = fmt.Sprintf("pod:%s:%s:%s:%d", ep.TargetName, endpoints.Name, a.IP, p.Port)
+						ep.ID = fmt.Sprintf("pod:%s:%s:%s:%s:%d", ep.TargetName, endpoints.Name, p.Name, a.IP, p.Port)
 					} else {
-						ep.ID = fmt.Sprintf("ept:%s:%s:%d", endpoints.Name, a.IP, p.Port)
+						ep.ID = fmt.Sprintf("ept:%s:%s:%s:%d", endpoints.Name, p.Name, a.IP, p.Port)
 					}
 				} else {
 					ep.TargetName = a.IP
-					ep.ID = fmt.Sprintf("ept:%s:%s:%d", endpoints.Name, a.IP, p.Port)
+					ep.ID = fmt.Sprintf("ept:%s:%s:%s:%d", endpoints.Name, p.Name, a.IP, p.Port)
 				}
 
 				// IdHash contains an obfuscated internal IP address

pkg/router/template/plugin_test.go

@@ -270,14 +270,14 @@ func TestHandleEndpoints(t *testing.T) {
 				},
 				Subsets: []kapi.EndpointSubset{{
 					Addresses: []kapi.EndpointAddress{{IP: "1.1.1.1"}},
-					Ports:     []kapi.EndpointPort{{Port: 345}},
+					Ports:     []kapi.EndpointPort{{Port: 345, Name: "port"}},
 				}}, //not specifying a port to force the port 80 assumption
 			},
 			expectedServiceUnit: &ServiceUnit{
 				Name: "foo/test", //service name from kapi.endpoints object
 				EndpointTable: []Endpoint{
 					{
-						ID:   "ept:test:1.1.1.1:345",
+						ID:   "ept:test:port:1.1.1.1:345",
 						IP:   "1.1.1.1",
 						Port: "345",
 					},
@@ -294,14 +294,14 @@ func TestHandleEndpoints(t *testing.T) {
 				},
 				Subsets: []kapi.EndpointSubset{{
 					Addresses: []kapi.EndpointAddress{{IP: "2.2.2.2", TargetRef: &kapi.ObjectReference{Kind: "Pod", Name: "pod-1"}}},
-					Ports:     []kapi.EndpointPort{{Port: 8080}},
+					Ports:     []kapi.EndpointPort{{Port: 8080, Name: "port"}},
 				}},
 			},
 			expectedServiceUnit: &ServiceUnit{
 				Name: "foo/test",
 				EndpointTable: []Endpoint{
 					{
-						ID:   "pod:pod-1:test:2.2.2.2:8080",
+						ID:   "pod:pod-1:test:port:2.2.2.2:8080",
 						IP:   "2.2.2.2",
 						Port: "8080",
 					},
@@ -375,16 +375,16 @@ func TestHandleTCPEndpoints(t *testing.T) {
 				Subsets: []kapi.EndpointSubset{{
 					Addresses: []kapi.EndpointAddress{{IP: "1.1.1.1", TargetRef: &kapi.ObjectReference{Kind: "Pod", Name: "pod-1"}}},
 					Ports: []kapi.EndpointPort{
-						{Port: 345},
-						{Port: 346, Protocol: kapi.ProtocolUDP},
+						{Port: 345, Name: "tcp"},
+						{Port: 346, Protocol: kapi.ProtocolUDP, Name: "udp"},
 					},
 				}}, //not specifying a port to force the port 80 assumption
 			},
 			expectedServiceUnit: &ServiceUnit{
 				Name: "foo/test", //service name from kapi.endpoints object
 				EndpointTable: []Endpoint{
 					{
-						ID:   "pod:pod-1:test:1.1.1.1:345",
+						ID:   "pod:pod-1:test:tcp:1.1.1.1:345",
 						IP:   "1.1.1.1",
 						Port: "345",
 					},
@@ -402,16 +402,16 @@ func TestHandleTCPEndpoints(t *testing.T) {
 				Subsets: []kapi.EndpointSubset{{
 					Addresses: []kapi.EndpointAddress{{IP: "2.2.2.2"}},
 					Ports: []kapi.EndpointPort{
-						{Port: 8080},
-						{Port: 8081, Protocol: kapi.ProtocolUDP},
+						{Port: 8080, Name: "tcp"},
+						{Port: 8081, Protocol: kapi.ProtocolUDP, Name: "udp"},
 					},
 				}},
 			},
 			expectedServiceUnit: &ServiceUnit{
 				Name: "foo/test",
 				EndpointTable: []Endpoint{
 					{
-						ID:   "ept:test:2.2.2.2:8080",
+						ID:   "ept:test:tcp:2.2.2.2:8080",
 						IP:   "2.2.2.2",
 						Port: "8080",
 					},

pkg/router/template/router_test.go

@@ -770,3 +770,130 @@ func TestFilterNamespaces(t *testing.T) {
 		}
 	}
 }
+
+// TestCalculateServiceWeights tests calculating the service
+// endpoint weights
+func TestCalculateServiceWeights(t *testing.T) {
+	router := NewFakeTemplateRouter()
+
+	suKey1 := ServiceUnitKey("ns/svc1")
+	suKey2 := ServiceUnitKey("ns/svc2")
+	ep1 := Endpoint{
+		ID:     "ep1",
+		IP:     "ip",
+		Port:   "port",
+		IdHash: fmt.Sprintf("%x", md5.Sum([]byte("ep1ipport"))),
+	}
+	ep2 := Endpoint{
+		ID:     "ep2",
+		IP:     "ip",
+		Port:   "port",
+		IdHash: fmt.Sprintf("%x", md5.Sum([]byte("ep2ipport"))),
+	}
+	ep3 := Endpoint{
+		ID:     "ep3",
+		IP:     "ip",
+		Port:   "port",
+		IdHash: fmt.Sprintf("%x", md5.Sum([]byte("ep3ipport"))),
+	}
+
+	testCases := []struct {
+		name            string
+		serviceUnits    map[ServiceUnitKey][]Endpoint
+		serviceWeights  map[ServiceUnitKey]int32
+		expectedWeights map[ServiceUnitKey]int32
+	}{
+		{
+			name: "equally weighted services with same number of endpoints",
+			serviceUnits: map[ServiceUnitKey][]Endpoint{
+				suKey1: {ep1},
+				suKey2: {ep2},
+			},
+			serviceWeights: map[ServiceUnitKey]int32{
+				suKey1: 50,
+				suKey2: 50,
+			},
+			expectedWeights: map[ServiceUnitKey]int32{
+				suKey1: 256,
+				suKey2: 256,
+			},
+		},
+		{
+			name: "unequally weighted services with same number of endpoints",
+			serviceUnits: map[ServiceUnitKey][]Endpoint{
+				suKey1: {ep1},
+				suKey2: {ep2},
+			},
+			serviceWeights: map[ServiceUnitKey]int32{
+				suKey1: 25,
+				suKey2: 75,
+			},
+			expectedWeights: map[ServiceUnitKey]int32{
+				suKey1: 85,
+				suKey2: 256,
+			},
+		},
+		{
+			name: "services with equal weights and a different number of endpoints",
+			serviceUnits: map[ServiceUnitKey][]Endpoint{
+				suKey1: {ep1, ep2},
+				suKey2: {ep3},
+			},
+			serviceWeights: map[ServiceUnitKey]int32{
+				suKey1: 50,
+				suKey2: 50,
+			},
+			expectedWeights: map[ServiceUnitKey]int32{
+				suKey1: 128,
+				suKey2: 256,
+			},
+		},
+		{
+			name: "services with unequal weights and a different number of endpoints",
+			serviceUnits: map[ServiceUnitKey][]Endpoint{
+				suKey1: {ep1, ep2},
+				suKey2: {ep3},
+			},
+			serviceWeights: map[ServiceUnitKey]int32{
+				suKey1: 20,
+				suKey2: 60,
+			},
+			expectedWeights: map[ServiceUnitKey]int32{
+				suKey1: 42,
+				suKey2: 256,
+			},
+		},
+		{
+			name: "services with equal weights and a different number of endpoints, one of which is common",
+			serviceUnits: map[ServiceUnitKey][]Endpoint{
+				suKey1: {ep1, ep2},
+				suKey2: {ep2},
+			},
+			serviceWeights: map[ServiceUnitKey]int32{
+				suKey1: 50,
+				suKey2: 50,
+			},
+			expectedWeights: map[ServiceUnitKey]int32{
+				suKey1: 128,
+				suKey2: 256,
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		for suKey, eps := range tc.serviceUnits {
+			router.CreateServiceUnit(suKey)
+			router.AddEndpoints(suKey, eps)
+		}
+		endpointWeights := router.calculateServiceWeights(tc.serviceWeights)
+		if !reflect.DeepEqual(endpointWeights, tc.expectedWeights) {
+			t.Errorf("test %s: expected endpointWeights to be %v, got %v", tc.name, tc.expectedWeights, endpointWeights)
+		}
+		// Remove endpoints and service units so the same sample template router
+		// can be re-used.
+		for suKey := range tc.serviceUnits {
+			router.DeleteEndpoints(suKey)
+			router.DeleteServiceUnit(suKey)
+		}
+	}
+}

pkg/router/template/template_helper.go

@@ -123,7 +123,7 @@ func genCertificateHostName(hostname string, wildcard bool) string {
 // processEndpointsForAlias returns the list of endpoints for the given route's service
 // action argument further processes the list e.g. shuffle
 // The default action is in-order traversal of internal data structure that stores
-//   the endpoints (does not change the return order if the data structure did not mutate)
+// the endpoints (does not change the return order if the data structure did not mutate)
 func processEndpointsForAlias(alias ServiceAliasConfig, svc ServiceUnit, action string) []Endpoint {
 	endpoints := endpointsForAlias(alias, svc)
 	if strings.ToLower(action) == "shuffle" {

pkg/router/template/template_helper_test.go

@@ -1,6 +1,7 @@
 package templaterouter
 
 import (
+	"crypto/md5"
 	"fmt"
 	"io/ioutil"
 	"reflect"
@@ -701,3 +702,68 @@ func TestGetPrimaryAliasKey(t *testing.T) {
 		}
 	}
 }
+
+func TestProcessEndpointsForAlias(t *testing.T) {
+	router := NewFakeTemplateRouter()
+	alias := buildServiceAliasConfig("api-route", "stg", "api-stg.127.0.0.1.nip.io", "", routev1.TLSTerminationEdge, routev1.InsecureEdgeTerminationPolicyRedirect, false)
+	suKey := ServiceUnitKey("stg/svc")
+	router.CreateServiceUnit(suKey)
+	ep1 := Endpoint{
+		ID:     "ep1",
+		IP:     "ip",
+		Port:   "foo",
+		IdHash: fmt.Sprintf("%x", md5.Sum([]byte("ep1ipport"))),
+	}
+	ep2 := Endpoint{
+		ID:     "ep2",
+		IP:     "ip",
+		Port:   "foo",
+		IdHash: fmt.Sprintf("%x", md5.Sum([]byte("ep2ipport"))),
+	}
+	ep3 := Endpoint{
+		ID:     "ep3",
+		IP:     "ip",
+		Port:   "bar",
+		IdHash: fmt.Sprintf("%x", md5.Sum([]byte("ep3ipport"))),
+	}
+
+	testCases := []struct {
+		name           string
+		preferPort     string
+		endpoints      []Endpoint
+		expectedLength int
+	}{
+		{
+			name:           "2 basic endpoints with same Port string",
+			preferPort:     "foo",
+			endpoints:      []Endpoint{ep1, ep2},
+			expectedLength: 2,
+		},
+		{
+			name:           "3 basic endpoints with different Port string",
+			preferPort:     "foo",
+			endpoints:      []Endpoint{ep1, ep2, ep3},
+			expectedLength: 2,
+		},
+	}
+
+	for _, tc := range testCases {
+		alias.PreferPort = tc.preferPort
+		endpointsCopy := make([]Endpoint, len(tc.endpoints))
+		for i := range tc.endpoints {
+			endpointsCopy[i] = tc.endpoints[i]
+		}
+		router.AddEndpoints(suKey, endpointsCopy)
+		svc, _ := router.FindServiceUnit(suKey)
+		endpoints := processEndpointsForAlias(alias, svc, "")
+		if len(endpoints) != tc.expectedLength {
+			t.Errorf("test %s: got wrong number of endpoints. Expected %d got %d", tc.name, tc.expectedLength, len(endpoints))
+		}
+		if len(tc.endpoints) == tc.expectedLength {
+			if !reflect.DeepEqual(tc.endpoints, endpoints) {
+				t.Errorf("test %s: endpoints out of order. Expected %v got %v", tc.name, tc.endpoints, endpoints)
+			}
+		}
+		router.DeleteEndpoints(suKey)
+	}
+}