chore(source): standardize sources with merge endpionts and deduplication

endpoint/endpoint.go

@@ -222,6 +222,7 @@ type EndpointKey struct {
 	RecordType    string
 	SetIdentifier string
 	RecordTTL     TTL
+	Target        string
 }
 
 type ObjectRef = events.ObjectReference

source/ambassador_host.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
 	"strings"
 
 	ambassador "github.com/datawire/ambassador/pkg/api/getambassador.io/v2"
@@ -181,11 +180,7 @@ func (sc *ambassadorHostSource) Endpoints(ctx context.Context) ([]*endpoint.Endp
 		endpoints = append(endpoints, hostEndpoints...)
 	}
 
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 // endpointsFromHost extracts the endpoints from a Host object

source/connector.go

@@ -71,7 +71,7 @@ func (cs *connectorSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, e
 
 	log.Debugf("Received endpoints: %#v", endpoints)
 
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (cs *connectorSource) AddEventHandler(_ context.Context, _ func()) {}

source/contour_httpproxy.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
 	"text/template"
 
 	projectcontour "github.com/projectcontour/contour/apis/projectcontour/v1"
@@ -167,11 +166,7 @@ func (sc *httpProxySource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, e
 		endpoints = append(endpoints, hpEndpoints...)
 	}
 
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (sc *httpProxySource) endpointsFromTemplate(httpProxy *projectcontour.HTTPProxy) ([]*endpoint.Endpoint, error) {

source/crd.go

@@ -230,7 +230,7 @@ func (cs *crdSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error
 		}
 	}
 
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (cs *crdSource) watch(ctx context.Context, opts *metav1.ListOptions) (watch.Interface, error) {

source/f5_transportserver.go

@@ -131,7 +131,7 @@ func (ts *f5TransportServerSource) Endpoints(_ context.Context) ([]*endpoint.End
 		return nil, err
 	}
 
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (ts *f5TransportServerSource) AddEventHandler(_ context.Context, handler func()) {

source/f5_virtualserver.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
 	"strings"
 
 	f5 "github.com/F5Networks/k8s-bigip-ctlr/v2/config/apis/cis/v1"
@@ -131,12 +130,7 @@ func (vs *f5VirtualServerSource) Endpoints(_ context.Context) ([]*endpoint.Endpo
 		return nil, err
 	}
 
-	// Sort endpoints
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (vs *f5VirtualServerSource) AddEventHandler(_ context.Context, handler func()) {

source/f5_virtualserver_test.go

@@ -39,7 +39,6 @@ const defaultF5VirtualServerNamespace = "virtualserver"
 
 func TestF5VirtualServerEndpoints(t *testing.T) {
 	t.Parallel()
-
 	tests := []struct {
 		name             string
 		annotationFilter string
@@ -349,7 +348,7 @@ func TestF5VirtualServerEndpoints(t *testing.T) {
 			},
 			expected: []*endpoint.Endpoint{
 				{
-					DNSName:    "www.example.com",
+					DNSName:    "alias1.example.com",
 					Targets:    []string{"192.168.1.100"},
 					RecordType: endpoint.RecordTypeA,
 					RecordTTL:  0,
@@ -358,7 +357,7 @@ func TestF5VirtualServerEndpoints(t *testing.T) {
 					},
 				},
 				{
-					DNSName:    "alias1.example.com",
+					DNSName:    "alias2.example.com",
 					Targets:    []string{"192.168.1.100"},
 					RecordType: endpoint.RecordTypeA,
 					RecordTTL:  0,
@@ -367,7 +366,7 @@ func TestF5VirtualServerEndpoints(t *testing.T) {
 					},
 				},
 				{
-					DNSName:    "alias2.example.com",
+					DNSName:    "www.example.com",
 					Targets:    []string{"192.168.1.100"},
 					RecordType: endpoint.RecordTypeA,
 					RecordTTL:  0,
@@ -600,7 +599,7 @@ func TestF5VirtualServerEndpoints(t *testing.T) {
 			endpoints, err := source.Endpoints(context.Background())
 			require.NoError(t, err)
 			assert.Len(t, endpoints, len(tc.expected))
-			assert.Equal(t, tc.expected, endpoints)
+			validateEndpoints(t, endpoints, tc.expected)
 		})
 	}
 }

source/fake.go

@@ -75,7 +75,7 @@ func (sc *fakeSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, error)
 		endpoints[i] = sc.generateEndpoint()
 	}
 
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (sc *fakeSource) generateEndpoint() *endpoint.Endpoint {

source/gateway.go

@@ -291,7 +291,7 @@ func (src *gatewayRouteSource) Endpoints(_ context.Context) ([]*endpoint.Endpoin
 
 		endpoints = append(endpoints, routeEndpoints...)
 	}
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func namespacedName(namespace, name string) types.NamespacedName {

source/gloo_proxy.go

@@ -236,7 +236,7 @@ func (gs *glooSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, error)
 			endpoints = append(endpoints, proxyEndpoints...)
 		}
 	}
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (gs *glooSource) generateEndpointsFromProxy(proxy *proxy, targets endpoint.Targets) ([]*endpoint.Endpoint, error) {

source/ingress.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
 	"strings"
 	"text/template"
 
@@ -179,11 +178,7 @@ func (sc *ingressSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, err
 		endpoints = append(endpoints, ingEndpoints...)
 	}
 
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (sc *ingressSource) endpointsFromTemplate(ing *networkv1.Ingress) ([]*endpoint.Endpoint, error) {

source/ingress_fqdn_test.go

@@ -166,7 +166,6 @@ func TestIngressSourceFqdnTemplatingExamples(t *testing.T) {
 			fqdnTemplate: `{{ range .Status.LoadBalancer.Ingress }}{{ if contains .Hostname "nip.io" }}example.org{{end}}{{end}}`,
 			expected: []*endpoint.Endpoint{
 				{DNSName: "example.org", RecordType: endpoint.RecordTypeCNAME, Targets: endpoint.Targets{"10.200.130.84.nip.io"}},
-				{DNSName: "example.org", RecordType: endpoint.RecordTypeCNAME, Targets: endpoint.Targets{"10.200.130.84.nip.io"}},
 			},
 		},
 		{

source/istio_gateway.go

@@ -19,7 +19,6 @@ package source
 import (
 	"context"
 	"fmt"
-	"sort"
 	"strings"
 	"text/template"
 
@@ -196,12 +195,7 @@ func (sc *gatewaySource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, e
 		endpoints = append(endpoints, gwEndpoints...)
 	}
 
-	// TODO: sort on endpoint creation
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 // AddEventHandler adds an event handler that should be triggered if the watched Istio Gateway changes.

source/istio_gateway_test.go

@@ -1880,7 +1880,6 @@ func TestSingleGatewayMultipleServicesPointingToSameLoadBalancer(t *testing.T) {
 
 	validateEndpoints(t, got, []*endpoint.Endpoint{
 		endpoint.NewEndpoint("example.org", endpoint.RecordTypeA, "34.66.66.77").WithLabel(endpoint.ResourceLabelKey, "gateway/argocd/argocd"),
-		endpoint.NewEndpoint("example.org", endpoint.RecordTypeA, "34.66.66.77").WithLabel(endpoint.ResourceLabelKey, "gateway/argocd/argocd"),
 	})
 }
 

source/istio_virtualservice.go

@@ -21,7 +21,6 @@ import (
 	"context"
 	"fmt"
 	"slices"
-	"sort"
 	"strings"
 	"text/template"
 
@@ -187,12 +186,7 @@ func (sc *virtualServiceSource) Endpoints(ctx context.Context) ([]*endpoint.Endp
 		endpoints = append(endpoints, gwEndpoints...)
 	}
 
-	// TODO: sort on endpoint creation
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 // AddEventHandler adds an event handler that should be triggered if the watched Istio VirtualService changes.

source/kong_tcpingress.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
 
 	log "github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
@@ -160,11 +159,7 @@ func (sc *kongTCPIngressSource) Endpoints(_ context.Context) ([]*endpoint.Endpoi
 		endpoints = append(endpoints, ingressEndpoints...)
 	}
 
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 // endpointsFromTCPIngress extracts the endpoints from a TCPIngress object

source/kong_tcpingress_test.go

@@ -374,8 +374,7 @@ func TestKongTCPIngressEndpoints(t *testing.T) {
 
 			endpoints, err := source.Endpoints(context.Background())
 			assert.NoError(t, err)
-			assert.Len(t, endpoints, len(ti.expected))
-			assert.Equal(t, ti.expected, endpoints)
+			validateEndpoints(t, endpoints, ti.expected)
 		})
 	}
 }

source/node.go

@@ -198,7 +198,7 @@ func (ns *nodeSource) endpointsForDNSNames(node *v1.Node, dnsNames []string) ([]
 		}
 	}
 
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 // nodeAddress returns the node's externalIP and if that's not found, the node's internalIP

source/openshift_route.go

@@ -19,7 +19,6 @@ package source
 import (
 	"context"
 	"fmt"
-	"sort"
 	"text/template"
 	"time"
 
@@ -159,11 +158,7 @@ func (ors *ocpRouteSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, e
 		endpoints = append(endpoints, orEndpoints...)
 	}
 
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (ors *ocpRouteSource) endpointsFromTemplate(ocpRoute *routev1.Route) ([]*endpoint.Endpoint, error) {

source/openshift_route_fqdn_test.go

@@ -209,7 +209,6 @@ func TestOpenShiftFqdnTemplatingExamples(t *testing.T) {
 			title:        "get canonical hostnames for admitted routes",
 			fqdnTemplate: "{{ $name := .Name }}{{ range $ingress := .Status.Ingress }}{{ range $ingress.Conditions }}{{ if and (eq .Type \"Admitted\") (eq .Status \"True\") }}{{ $ingress.Host }},{{ end }}{{ end }}{{ end }}",
 			expected: []*endpoint.Endpoint{
-				{DNSName: "cluster.example.org", RecordType: endpoint.RecordTypeCNAME, Targets: endpoint.Targets{"router-dmz.apps.dmz.example.com"}},
 				{DNSName: "cluster.example.org", RecordType: endpoint.RecordTypeCNAME, Targets: endpoint.Targets{"router-dmz.apps.dmz.example.com"}},
 				{DNSName: "apps.example.org", RecordType: endpoint.RecordTypeCNAME, Targets: endpoint.Targets{"router-dmz.apps.dmz.example.com"}},
 			},

source/service.go

@@ -330,7 +330,7 @@ func (sc *serviceSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, err
 		})
 	}
 
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 // extractHeadlessEndpoints extracts endpoints from a headless service using the "Endpoints" Kubernetes API resource

source/skipper_routegroup.go

@@ -27,7 +27,6 @@ import (
 	"net/http"
 	"net/url"
 	"os"
-	"sort"
 	"strings"
 	"sync"
 	"text/template"
@@ -288,11 +287,7 @@ func (sc *routeGroupSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint,
 		endpoints = append(endpoints, eps...)
 	}
 
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 func (sc *routeGroupSource) endpointsFromTemplate(rg *routeGroup) ([]*endpoint.Endpoint, error) {

source/traefik_proxy.go

@@ -21,7 +21,6 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
-	"sort"
 	"strings"
 
 	log "github.com/sirupsen/logrus"
@@ -209,11 +208,7 @@ func (ts *traefikSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, err
 		endpoints = append(endpoints, oldIngressRouteUdpEndpoints...)
 	}
 
-	for _, ep := range endpoints {
-		sort.Sort(ep.Targets)
-	}
-
-	return endpoints, nil
+	return MergeEndpoints(endpoints), nil
 }
 
 // ingressRouteEndpoints extracts endpoints from all IngressRoute objects