chore(source): standardize sources with merge endpionts and deduplicate targets

ivankatliarchuk · ivankatliarchuk · commit 8e4bf8eb7970 · 2026-02-07T20:49:09.000Z
Signed-off-by: ivan katliarchuk &lt;ivan.katliarchuk@gmail.com&gt;
diff --git a/endpoint/endpoint.go b/endpoint/endpoint.go
@@ -222,6 +222,7 @@ type EndpointKey struct {
 	RecordType    string
 	SetIdentifier string
 	RecordTTL     TTL
+	Target        string
 }
 
 type ObjectRef = events.ObjectReference
diff --git a/source/ambassador_host.go b/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
diff --git a/source/connector.go b/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()) {}
diff --git a/source/contour_httpproxy.go b/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) {
diff --git a/source/crd.go b/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) {
diff --git a/source/f5_transportserver.go b/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()) {
diff --git a/source/f5_virtualserver.go b/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()) {
diff --git a/source/f5_virtualserver_test.go b/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)
 		})
 	}
 }
diff --git a/source/fake.go b/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 {
diff --git a/source/gateway.go b/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 {
diff --git a/source/gloo_proxy.go b/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) {
diff --git a/source/ingress.go b/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) {
diff --git a/source/ingress_fqdn_test.go b/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"}},
 			},
 		},
 		{
diff --git a/source/istio_gateway.go b/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.
diff --git a/source/istio_gateway_test.go b/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"),
 	})
 }
 
diff --git a/source/istio_virtualservice.go b/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.
diff --git a/source/kong_tcpingress.go b/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
diff --git a/source/kong_tcpingress_test.go b/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)
 		})
 	}
 }
diff --git a/source/node.go b/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
diff --git a/source/openshift_route.go b/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) {
diff --git a/source/openshift_route_fqdn_test.go b/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"}},
 			},
diff --git a/source/service.go b/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
diff --git a/source/skipper_routegroup.go b/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) {
diff --git a/source/traefik_proxy.go b/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
diff --git a/source/traefik_proxy_test.go b/source/traefik_proxy_test.go
diff --git a/source/utils.go b/source/utils.go
diff --git a/source/utils_test.go b/source/utils_test.go

Original file line number	Diff line number	Diff line change
`@@ -222,6 +222,7 @@ type EndpointKey struct {`
`222`	`222`	`RecordType string`
`223`	`223`	`SetIdentifier string`
`224`	`224`	`RecordTTL TTL`
	`225`	`+ Target string`
`225`	`226`	`}`
`226`	`227`
`227`	`228`	`type ObjectRef = events.ObjectReference`
Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ func (cs connectorSource) Endpoints(_ context.Context) ([]endpoint.Endpoint, e`
`71`	`71`
`72`	`72`	`log.Debugf("Received endpoints: %#v", endpoints)`
`73`	`73`
`74`		`- return endpoints, nil`
	`74`	`+ return MergeEndpoints(endpoints), nil`
`75`	`75`	`}`
`76`	`76`
`77`	`77`	`func (cs *connectorSource) AddEventHandler(_ context.Context, _ func()) {}`
Original file line number	Diff line number	Diff line change
`@@ -230,7 +230,7 @@ func (cs crdSource) Endpoints(ctx context.Context) ([]endpoint.Endpoint, error`
`230`	`230`	`}`
`231`	`231`	`}`
`232`	`232`
`233`		`- return endpoints, nil`
	`233`	`+ return MergeEndpoints(endpoints), nil`
`234`	`234`	`}`
`235`	`235`
`236`	`236`	`func (cs crdSource) watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error) {`
Original file line number	Diff line number	Diff line change
`@@ -131,7 +131,7 @@ func (ts f5TransportServerSource) Endpoints(_ context.Context) ([]endpoint.End`
`131`	`131`	`return nil, err`
`132`	`132`	`}`
`133`	`133`
`134`		`- return endpoints, nil`
	`134`	`+ return MergeEndpoints(endpoints), nil`
`135`	`135`	`}`
`136`	`136`
`137`	`137`	`func (ts *f5TransportServerSource) AddEventHandler(_ context.Context, handler func()) {`
Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ func (sc fakeSource) Endpoints(_ context.Context) ([]endpoint.Endpoint, error)`
`75`	`75`	`endpoints[i] = sc.generateEndpoint()`
`76`	`76`	`}`
`77`	`77`
`78`		`- return endpoints, nil`
	`78`	`+ return MergeEndpoints(endpoints), nil`
`79`	`79`	`}`
`80`	`80`
`81`	`81`	`func (sc fakeSource) generateEndpoint() endpoint.Endpoint {`
Original file line number	Diff line number	Diff line change
`@@ -291,7 +291,7 @@ func (src gatewayRouteSource) Endpoints(_ context.Context) ([]endpoint.Endpoin`
`291`	`291`
`292`	`292`	`endpoints = append(endpoints, routeEndpoints...)`
`293`	`293`	`}`
`294`		`- return endpoints, nil`
	`294`	`+ return MergeEndpoints(endpoints), nil`
`295`	`295`	`}`
`296`	`296`
`297`	`297`	`func namespacedName(namespace, name string) types.NamespacedName {`