refactor(annotation): rationalize filtering

source/ambassador_host.go

@@ -138,7 +138,7 @@ func (sc *ambassadorHostSource) Endpoints(ctx context.Context) ([]*endpoint.Endp
 	}
 
 	// Filter Ambassador Hosts
-	ambassadorHosts, err = sc.filterByAnnotations(ambassadorHosts)
+	ambassadorHosts, err = annotations.Filter(ambassadorHosts, sc.annotationFilter)
 	if err != nil {
 		return nil, fmt.Errorf("failed to filter Ambassador Hosts by annotation: %w", err)
 	}
@@ -290,28 +290,3 @@ func newUnstructuredConverter() (*unstructuredConverter, error) {
 
 	return uc, nil
 }
-
-// Filter a list of Ambassador Host Resources to only return the ones that
-// contain the required External-DNS annotation filter
-func (sc *ambassadorHostSource) filterByAnnotations(ambassadorHosts []*ambassador.Host) ([]*ambassador.Host, error) {
-	selector, err := annotations.ParseFilter(sc.annotationFilter)
-	if err != nil {
-		return nil, err
-	}
-
-	// empty filter returns original list of Ambassador Hosts
-	if selector.Empty() {
-		return ambassadorHosts, nil
-	}
-
-	// Return a filtered list of Ambassador Hosts
-	filteredList := []*ambassador.Host{}
-	for _, host := range ambassadorHosts {
-		// include Ambassador Host if its annotations match the annotation filter
-		if selector.Matches(labels.Set(host.Annotations)) {
-			filteredList = append(filteredList, host)
-		}
-	}
-
-	return filteredList, nil
-}

source/annotations/filter.go

@@ -0,0 +1,50 @@
+/*
+Copyright 2025 The Kubernetes 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 annotations
+
+import (
+	"strings"
+
+	log "github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+// AnnotatedObject represents any Kubernetes object with annotations
+type AnnotatedObject interface {
+	GetAnnotations() map[string]string
+}
+
+// Filter filters a slice of objects by annotation selector.
+// Returns all items if annotationFilter is empty.
+func Filter[T AnnotatedObject](items []T, filter string) ([]T, error) {
+	if filter == "" || strings.TrimSpace(filter) == "" {
+		return items, nil
+	}
+	selector, err := ParseFilter(filter)
+	if err != nil {
+		return nil, err
+	}
+	if selector.Empty() {
+		return items, nil
+	}
+
+	filtered := make([]T, 0, len(items))
+	for _, item := range items {
+		if selector.Matches(labels.Set(item.GetAnnotations())) {
+			filtered = append(filtered, item)
+		}
+	}
+	log.Debugf("filtered '%d' services out of '%d' with annotation filter '%s'", len(filtered), len(items), filter)
+	return filtered, nil
+}

source/annotations/filter_test.go

@@ -0,0 +1,136 @@
+/*
+Copyright 2025 The Kubernetes 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 annotations
+
+import (
+	"testing"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"sigs.k8s.io/external-dns/internal/testutils"
+)
+
+// Mock object implementing AnnotatedObject
+type mockObj struct {
+	annotations map[string]string
+}
+
+func (m mockObj) GetAnnotations() map[string]string {
+	return m.annotations
+}
+
+func TestFilter(t *testing.T) {
+	tests := []struct {
+		name        string
+		items       []mockObj
+		filter      string
+		expected    []mockObj
+		expectError bool
+	}{
+		{
+			name: "Empty filter returns all",
+			items: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"baz": "qux"}},
+			},
+			filter: "",
+			expected: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"baz": "qux"}},
+			},
+		},
+		{
+			name: "Matching items",
+			items: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"foo": "baz"}},
+			},
+			filter: "foo=bar",
+			expected: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+			},
+		},
+		{
+			name: "No matching items",
+			items: []mockObj{
+				{annotations: map[string]string{"foo": "baz"}},
+			},
+			filter:   "foo=bar",
+			expected: []mockObj{},
+		},
+		{
+			name: "Whitespace filter returns all",
+			items: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"baz": "qux"}},
+			},
+			filter: "   ",
+			expected: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"baz": "qux"}},
+			},
+		},
+		{
+			name: "empty filter returns all",
+			items: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"baz": "qux"}},
+			},
+			filter: "",
+			expected: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"baz": "qux"}},
+			},
+		},
+		{
+			name: "invalid filter returns error",
+			items: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"baz": "qux"}},
+			},
+			filter: "=invalid",
+			expected: []mockObj{
+				{annotations: map[string]string{"foo": "bar"}},
+				{annotations: map[string]string{"baz": "qux"}},
+			},
+			expectError: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := Filter(tt.items, tt.filter)
+			if tt.expectError {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tt.expected, result)
+			}
+		})
+	}
+}
+
+func TestFilter_LogOutput(t *testing.T) {
+	hook := testutils.LogsUnderTestWithLogLevel(log.DebugLevel, t)
+
+	items := []mockObj{
+		{annotations: map[string]string{"foo": "bar"}},
+		{annotations: map[string]string{"foo": "baz"}},
+	}
+	filter := "foo=bar"
+	_, _ = Filter(items, filter)
+
+	testutils.TestHelperLogContains("filtered '1' services out of '2' with annotation filter 'foo=bar'", hook, t)
+}

source/contour_httpproxy.go

@@ -127,7 +127,7 @@ func (sc *httpProxySource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint,
 		httpProxies = append(httpProxies, hpConverted)
 	}
 
-	httpProxies, err = sc.filterByAnnotations(httpProxies)
+	httpProxies, err = annotations.Filter(httpProxies, sc.annotationFilter)
 	if err != nil {
 		return nil, fmt.Errorf("failed to filter HTTPProxies: %w", err)
 	}
@@ -209,30 +209,6 @@ func (sc *httpProxySource) endpointsFromTemplate(httpProxy *projectcontour.HTTPP
 	return endpoints, nil
 }
 
-// filterByAnnotations filters a list of configs by a given annotation selector.
-func (sc *httpProxySource) filterByAnnotations(httpProxies []*projectcontour.HTTPProxy) ([]*projectcontour.HTTPProxy, error) {
-	selector, err := annotations.ParseFilter(sc.annotationFilter)
-	if err != nil {
-		return nil, err
-	}
-
-	// empty filter returns original list
-	if selector.Empty() {
-		return httpProxies, nil
-	}
-
-	var filteredList []*projectcontour.HTTPProxy
-
-	for _, httpProxy := range httpProxies {
-		// include HTTPProxy if its annotations match the selector
-		if selector.Matches(labels.Set(httpProxy.Annotations)) {
-			filteredList = append(filteredList, httpProxy)
-		}
-	}
-
-	return filteredList, nil
-}
-
 // endpointsFromHTTPProxyConfig extracts the endpoints from a Contour HTTPProxy object
 func (sc *httpProxySource) endpointsFromHTTPProxy(httpProxy *projectcontour.HTTPProxy) ([]*endpoint.Endpoint, error) {
 	resource := fmt.Sprintf("HTTPProxy/%s/%s", httpProxy.Namespace, httpProxy.Name)

source/crd.go

@@ -111,7 +111,12 @@ func NewCRDClientForAPIVersionKind(client kubernetes.Interface, kubeConfig, apiS
 }
 
 // NewCRDSource creates a new crdSource with the given config.
-func NewCRDSource(crdClient rest.Interface, namespace, kind string, annotationFilter string, labelSelector labels.Selector, scheme *runtime.Scheme, startInformer bool) (Source, error) {
+func NewCRDSource(
+	crdClient rest.Interface,
+	namespace, kind, annotationFilter string,
+	labelSelector labels.Selector,
+	scheme *runtime.Scheme,
+	startInformer bool) (Source, error) {
 	sourceCrd := crdSource{
 		crdResource:      strings.ToLower(kind) + "s",
 		namespace:        namespace,
@@ -174,12 +179,17 @@ func (cs *crdSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error
 		return nil, err
 	}
 
-	result, err = cs.filterByAnnotations(result)
+	itemPtrs := make([]*apiv1alpha1.DNSEndpoint, len(result.Items))
+	for i := range result.Items {
+		itemPtrs[i] = &result.Items[i]
+	}
+
+	filtered, err := annotations.Filter(itemPtrs, cs.annotationFilter)
 	if err != nil {
 		return nil, err
 	}
 
-	for _, dnsEndpoint := range result.Items {
+	for _, dnsEndpoint := range filtered {
 		var crdEndpoints []*endpoint.Endpoint
 		for _, ep := range dnsEndpoint.Spec.Endpoints {
 			if (ep.RecordType == endpoint.RecordTypeCNAME || ep.RecordType == endpoint.RecordTypeA || ep.RecordType == endpoint.RecordTypeAAAA) && len(ep.Targets) < 1 {
@@ -214,7 +224,7 @@ func (cs *crdSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error
 
 		dnsEndpoint.Status.ObservedGeneration = dnsEndpoint.Generation
 		// Update the ObservedGeneration
-		_, err = cs.UpdateStatus(ctx, &dnsEndpoint)
+		_, err = cs.UpdateStatus(ctx, dnsEndpoint)
 		if err != nil {
 			log.Warnf("Could not update ObservedGeneration of the CRD: %v", err)
 		}
@@ -253,26 +263,3 @@ func (cs *crdSource) UpdateStatus(ctx context.Context, dnsEndpoint *apiv1alpha1.
 		Do(ctx).
 		Into(result)
 }
-
-// filterByAnnotations filters a list of dnsendpoints by a given annotation selector.
-func (cs *crdSource) filterByAnnotations(dnsendpoints *apiv1alpha1.DNSEndpointList) (*apiv1alpha1.DNSEndpointList, error) {
-	selector, err := annotations.ParseFilter(cs.annotationFilter)
-	if err != nil {
-		return nil, err
-	}
-	// empty filter returns original list
-	if selector.Empty() {
-		return dnsendpoints, nil
-	}
-
-	filteredList := apiv1alpha1.DNSEndpointList{}
-
-	for _, dnsendpoint := range dnsendpoints.Items {
-		// include dnsendpoint if its annotations match the selector
-		if selector.Matches(labels.Set(dnsendpoint.Annotations)) {
-			filteredList.Items = append(filteredList.Items, dnsendpoint)
-		}
-	}
-
-	return &filteredList, nil
-}

source/f5_transportserver.go

@@ -120,7 +120,7 @@ func (ts *f5TransportServerSource) Endpoints(ctx context.Context) ([]*endpoint.E
 		transportServers = append(transportServers, transportServer)
 	}
 
-	transportServers, err = ts.filterByAnnotations(transportServers)
+	transportServers, err = annotations.Filter(transportServers, ts.annotationFilter)
 	if err != nil {
 		return nil, fmt.Errorf("failed to filter TransportServers: %w", err)
 	}
@@ -183,30 +183,6 @@ func newTSUnstructuredConverter() (*unstructuredConverter, error) {
 	return uc, nil
 }
 
-// filterByAnnotations filters a list of TransportServers by a given annotation selector.
-func (ts *f5TransportServerSource) filterByAnnotations(transportServers []*f5.TransportServer) ([]*f5.TransportServer, error) {
-	selector, err := annotations.ParseFilter(ts.annotationFilter)
-	if err != nil {
-		return nil, err
-	}
-
-	// empty filter returns original list
-	if selector.Empty() {
-		return transportServers, nil
-	}
-
-	filteredList := []*f5.TransportServer{}
-
-	for _, ts := range transportServers {
-		// include TransportServer if its annotations match the selector
-		if selector.Matches(labels.Set(ts.Annotations)) {
-			filteredList = append(filteredList, ts)
-		}
-	}
-
-	return filteredList, nil
-}
-
 func hasValidTransportServerIP(vs *f5.TransportServer) bool {
 	normalizedAddress := strings.ToLower(vs.Status.VSAddress)
 	return normalizedAddress != "none" && normalizedAddress != ""