feat(source/pods): support for annotation and label filter (#5583)

* feat(source): pods added support for annotation filter and label selectors

* feat(source/pods): support for annotation and label filter

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

---------

Signed-off-by: ivan katliarchuk <ivan.katliarchuk@gmail.com>

docs/sources/about.md

@@ -5,26 +5,26 @@ A source in ExternalDNS defines where DNS records are discovered from within you
 ExternalDNS watches the specified sources for hostname information and uses it to create, update, or delete DNS records accordingly. Multiple sources can be configured simultaneously to support diverse environments.
 
 | Source                                  | Resources                                                                     | annotation-filter | label-filter |
-| --------------------------------------- | ----------------------------------------------------------------------------- | ----------------- | ------------ |
-| ambassador-host                         | Host.getambassador.io                                                         | Yes               | Yes          |
+|-----------------------------------------|-------------------------------------------------------------------------------|:-----------------:|:------------:|
+| ambassador-host                         | Host.getambassador.io                                                         |        Yes        |     Yes      |
 | connector                               |                                                                               |                   |              |
-| contour-httpproxy                       | HttpProxy.projectcontour.io                                                   | Yes               |              |
+| contour-httpproxy                       | HttpProxy.projectcontour.io                                                   |        Yes        |              |
 | cloudfoundry                            |                                                                               |                   |              |
-| [crd](crd.md)                           | DNSEndpoint.externaldns.k8s.io                                                | Yes               | Yes          |
-| [f5-virtualserver](f5-virtualserver.md) | VirtualServer.cis.f5.com                                                      | Yes               |              |
-| [gateway-grpcroute](gateway.md)         | GRPCRoute.gateway.networking.k8s.io                                           | Yes               | Yes          |
-| [gateway-httproute](gateway.md)         | HTTPRoute.gateway.networking.k8s.io                                           | Yes               | Yes          |
-| [gateway-tcproute](gateway.md)          | TCPRoute.gateway.networking.k8s.io                                            | Yes               | Yes          |
-| [gateway-tlsroute](gateway.md)          | TLSRoute.gateway.networking.k8s.io                                            | Yes               | Yes          |
-| [gateway-udproute](gateway.md)          | UDPRoute.gateway.networking.k8s.io                                            | Yes               | Yes          |
+| [crd](crd.md)                           | DNSEndpoint.externaldns.k8s.io                                                |        Yes        |     Yes      |
+| [f5-virtualserver](f5-virtualserver.md) | VirtualServer.cis.f5.com                                                      |        Yes        |              |
+| [gateway-grpcroute](gateway.md)         | GRPCRoute.gateway.networking.k8s.io                                           |        Yes        |     Yes      |
+| [gateway-httproute](gateway.md)         | HTTPRoute.gateway.networking.k8s.io                                           |        Yes        |     Yes      |
+| [gateway-tcproute](gateway.md)          | TCPRoute.gateway.networking.k8s.io                                            |        Yes        |     Yes      |
+| [gateway-tlsroute](gateway.md)          | TLSRoute.gateway.networking.k8s.io                                            |        Yes        |     Yes      |
+| [gateway-udproute](gateway.md)          | UDPRoute.gateway.networking.k8s.io                                            |        Yes        |     Yes      |
 | [gloo-proxy](gloo-proxy.md)             | Proxy.gloo.solo.io                                                            |                   |              |
-| [ingress](ingress.md)                   | Ingress.networking.k8s.io                                                     | Yes               | Yes          |
-| [istio-gateway](istio.md)               | Gateway.networking.istio.io                                                   | Yes               |              |
-| [istio-virtualservice](istio.md)        | VirtualService.networking.istio.io                                            | Yes               |              |
-| [kong-tcpingress](kong.md)              | TCPIngress.configuration.konghq.com                                           | Yes               |              |
-| [node](nodes.md)                        | Node                                                                          | Yes               | Yes          |
-| [openshift-route](openshift.md)         | Route.route.openshift.io                                                      | Yes               | Yes          |
-| [pod](pod.md)                           | Pod                                                                           |                   |              |
-| [service](service.md)                   | Service                                                                       | Yes               | Yes          |
-| skipper-routegroup                      | RouteGroup.zalando.org                                                        | Yes               |              |
-| [traefik-proxy](traefik-proxy.md)       | IngressRoute.traefik.io IngressRouteTCP.traefik.io IngressRouteUDP.traefik.io | Yes               |              |
+| [ingress](ingress.md)                   | Ingress.networking.k8s.io                                                     |        Yes        |     Yes      |
+| [istio-gateway](istio.md)               | Gateway.networking.istio.io                                                   |        Yes        |              |
+| [istio-virtualservice](istio.md)        | VirtualService.networking.istio.io                                            |        Yes        |              |
+| [kong-tcpingress](kong.md)              | TCPIngress.configuration.konghq.com                                           |        Yes        |              |
+| [node](nodes.md)                        | Node                                                                          |        Yes        |     Yes      |
+| [openshift-route](openshift.md)         | Route.route.openshift.io                                                      |        Yes        |     Yes      |
+| [pod](pod.md)                           | Pod                                                                           |        Yes        |     Yes      |
+| [service](service.md)                   | Service                                                                       |        Yes        |     Yes      |
+| skipper-routegroup                      | RouteGroup.zalando.org                                                        |        Yes        |              |
+| [traefik-proxy](traefik-proxy.md)       | IngressRoute.traefik.io IngressRouteTCP.traefik.io IngressRouteUDP.traefik.io |        Yes        |              |

source/annotations/processors_test.go

@@ -47,6 +47,12 @@ func TestParseAnnotationFilter(t *testing.T) {
 			expectedSelector: labels.Set{}.AsSelector(),
 			expectError:      false,
 		},
+		{
+			name:             "wrong annotation filter",
+			annotationFilter: "=test",
+			expectedSelector: nil,
+			expectError:      true,
+		},
 	}
 
 	for _, tt := range tests {

source/informers/indexers.go

@@ -0,0 +1,113 @@
+/*
+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 informers
+
+import (
+	"fmt"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/cache"
+
+	"sigs.k8s.io/external-dns/source/annotations"
+)
+
+const (
+	IndexWithSelectors = "withSelectors"
+)
+
+type IndexSelectorOptions struct {
+	annotationFilter labels.Selector
+	labelSelector    labels.Selector
+}
+
+func IndexSelectorWithAnnotationFilter(input string) func(options *IndexSelectorOptions) {
+	return func(options *IndexSelectorOptions) {
+		if input == "" {
+			return
+		}
+		selector, err := annotations.ParseFilter(input)
+		if err != nil {
+			return
+		}
+		options.annotationFilter = selector
+	}
+}
+
+func IndexSelectorWithLabelSelector(input labels.Selector) func(options *IndexSelectorOptions) {
+	return func(options *IndexSelectorOptions) {
+		options.labelSelector = input
+	}
+}
+
+// IndexerWithOptions is a generic function that allows adding multiple indexers
+// to a SharedIndexInformer for a specific Kubernetes resource type T. It accepts
+// a variadic list of indexer functions, which define custom indexing logic.
+//
+// Each indexer function is applied to objects of type T, enabling flexible and
+// reusable indexing based on annotations, labels, or other criteria.
+//
+// Example usage:
+// err := IndexerWithOptions[*v1.Pod](
+//
+//	IndexSelectorWithAnnotationFilter("example-annotation"),
+//	IndexSelectorWithLabelSelector(labels.SelectorFromSet(labels.Set{"app": "my-app"})),
+//
+// )
+//
+// This function ensures type safety and simplifies the process of adding
+// custom indexers to informers.
+func IndexerWithOptions[T metav1.Object](optFns ...func(options *IndexSelectorOptions)) cache.Indexers {
+	options := IndexSelectorOptions{}
+	for _, fn := range optFns {
+		fn(&options)
+	}
+
+	return cache.Indexers{
+		IndexWithSelectors: func(obj interface{}) ([]string, error) {
+			entity, ok := obj.(T)
+			if !ok {
+				return nil, fmt.Errorf("object is not of type %T", new(T))
+			}
+
+			if options.annotationFilter != nil && !options.annotationFilter.Matches(labels.Set(entity.GetAnnotations())) {
+				return nil, nil
+			}
+
+			if options.labelSelector != nil && !options.labelSelector.Matches(labels.Set(entity.GetLabels())) {
+				return nil, nil
+			}
+			key := types.NamespacedName{Namespace: entity.GetNamespace(), Name: entity.GetName()}.String()
+			return []string{key}, nil
+		},
+	}
+}
+
+// GetByKey retrieves an object of type T (metav1.Object) from the given cache.Indexer by its key.
+// It returns the object and an error if the retrieval or type assertion fails.
+// If the object does not exist, it returns the zero value of T and nil.
+func GetByKey[T metav1.Object](indexer cache.Indexer, key string) (T, error) {
+	var entity T
+	obj, exists, err := indexer.GetByKey(key)
+	if err != nil || !exists {
+		return entity, err
+	}
+
+	entity, ok := obj.(T)
+	if !ok {
+		return entity, fmt.Errorf("object is not of type %T", new(T))
+	}
+	return entity, nil
+}

source/informers/indexers_test.go

@@ -0,0 +1,185 @@
+/*
+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 informers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+	"sigs.k8s.io/external-dns/source/annotations"
+)
+
+func TestIndexerWithOptions_FilterByAnnotation(t *testing.T) {
+	indexers := IndexerWithOptions[*unstructured.Unstructured](
+		IndexSelectorWithAnnotationFilter("example-annotation"),
+	)
+
+	obj := &unstructured.Unstructured{}
+	obj.SetAnnotations(map[string]string{"example-annotation": "value"})
+	obj.SetNamespace("default")
+	obj.SetName("test-object")
+
+	keys, err := indexers[IndexWithSelectors](obj)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"default/test-object"}, keys)
+}
+
+func TestIndexerWithOptions_FilterByLabel(t *testing.T) {
+	labelSelector := labels.SelectorFromSet(labels.Set{"app": "nginx"})
+	indexers := IndexerWithOptions[*corev1.Pod](
+		IndexSelectorWithLabelSelector(labelSelector),
+	)
+
+	obj := &corev1.Pod{}
+	obj.SetLabels(map[string]string{"app": "nginx"})
+	obj.SetNamespace("default")
+	obj.SetName("test-object")
+
+	keys, err := indexers[IndexWithSelectors](obj)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"default/test-object"}, keys)
+}
+
+func TestIndexerWithOptions_NoMatch(t *testing.T) {
+	labelSelector := labels.SelectorFromSet(labels.Set{"app": "nginx"})
+	indexers := IndexerWithOptions[*unstructured.Unstructured](
+		IndexSelectorWithLabelSelector(labelSelector),
+	)
+
+	obj := &unstructured.Unstructured{}
+	obj.SetLabels(map[string]string{"app": "apache"})
+	obj.SetNamespace("default")
+	obj.SetName("test-object")
+
+	keys, err := indexers[IndexWithSelectors](obj)
+	assert.NoError(t, err)
+	assert.Nil(t, keys)
+}
+
+func TestIndexerWithOptions_InvalidType(t *testing.T) {
+	indexers := IndexerWithOptions[*unstructured.Unstructured]()
+
+	obj := "invalid-object"
+
+	keys, err := indexers[IndexWithSelectors](obj)
+	assert.Error(t, err)
+	assert.Nil(t, keys)
+	assert.Contains(t, err.Error(), "object is not of type")
+}
+
+func TestIndexerWithOptions_EmptyOptions(t *testing.T) {
+	indexers := IndexerWithOptions[*unstructured.Unstructured]()
+
+	obj := &unstructured.Unstructured{}
+	obj.SetNamespace("default")
+	obj.SetName("test-object")
+
+	keys, err := indexers["withSelectors"](obj)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"default/test-object"}, keys)
+}
+
+func TestIndexerWithOptions_AnnotationFilterNoMatch(t *testing.T) {
+	indexers := IndexerWithOptions[*unstructured.Unstructured](
+		IndexSelectorWithAnnotationFilter("example-annotation=value"),
+	)
+
+	obj := &unstructured.Unstructured{}
+	obj.SetAnnotations(map[string]string{"other-annotation": "value"})
+	obj.SetNamespace("default")
+	obj.SetName("test-object")
+
+	keys, err := indexers[IndexWithSelectors](obj)
+	assert.NoError(t, err)
+	assert.Nil(t, keys)
+}
+
+func TestIndexSelectorWithAnnotationFilter(t *testing.T) {
+	tests := []struct {
+		name           string
+		input          string
+		expectedFilter labels.Selector
+	}{
+		{
+			name:           "valid input",
+			input:          "key=value",
+			expectedFilter: func() labels.Selector { s, _ := annotations.ParseFilter("key=value"); return s }(),
+		},
+		{
+			name:           "empty input",
+			input:          "",
+			expectedFilter: nil,
+		},
+		{
+			name:           "key only filter",
+			input:          "app",
+			expectedFilter: func() labels.Selector { s, _ := annotations.ParseFilter("app"); return s }(),
+		},
+		{
+			name:           "poisoned input",
+			input:          "=app",
+			expectedFilter: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			options := &IndexSelectorOptions{}
+			IndexSelectorWithAnnotationFilter(tt.input)(options)
+			assert.Equal(t, tt.expectedFilter, options.annotationFilter)
+		})
+	}
+}
+
+func TestGetByKey_ObjectExists(t *testing.T) {
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+	pod := &corev1.Pod{}
+	pod.SetNamespace("default")
+	pod.SetName("test-pod")
+
+	err := indexer.Add(pod)
+	assert.NoError(t, err)
+
+	result, err := GetByKey[*corev1.Pod](indexer, "default/test-pod")
+	assert.NoError(t, err)
+	assert.NotNil(t, result)
+	assert.Equal(t, "test-pod", result.GetName())
+}
+
+func TestGetByKey_ObjectDoesNotExist(t *testing.T) {
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+
+	result, err := GetByKey[*corev1.Pod](indexer, "default/non-existent-pod")
+	assert.NoError(t, err)
+	assert.Nil(t, result)
+}
+
+func TestGetByKey_TypeAssertionFailure(t *testing.T) {
+	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+	service := &corev1.Service{}
+	service.SetNamespace("default")
+	service.SetName("test-service")
+
+	err := indexer.Add(service)
+	assert.NoError(t, err)
+
+	result, err := GetByKey[*corev1.Pod](indexer, "default/test-service")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "object is not of type")
+	assert.Nil(t, result)
+}

source/pod.go

@@ -25,15 +25,15 @@ import (
 	log "github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
+
 	kubeinformers "k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 
-	"sigs.k8s.io/external-dns/source/fqdn"
-
 	"sigs.k8s.io/external-dns/endpoint"
 	"sigs.k8s.io/external-dns/source/annotations"
+	"sigs.k8s.io/external-dns/source/fqdn"
 	"sigs.k8s.io/external-dns/source/informers"
 )
 
@@ -60,11 +60,22 @@ func NewPodSource(
 	podSourceDomain string,
 	fqdnTemplate string,
 	combineFqdnAnnotation bool,
+	annotationFilter string,
+	labelSelector labels.Selector,
 ) (Source, error) {
 	informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, 0, kubeinformers.WithNamespace(namespace))
 	podInformer := informerFactory.Core().V1().Pods()
 	nodeInformer := informerFactory.Core().V1().Nodes()
 
+	err := podInformer.Informer().AddIndexers(informers.IndexerWithOptions[*corev1.Pod](
+		informers.IndexSelectorWithAnnotationFilter(annotationFilter),
+		informers.IndexSelectorWithLabelSelector(labelSelector),
+	))
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to add indexers to pod informer: %w", err)
+	}
+
 	_, _ = podInformer.Informer().AddEventHandler(
 		cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
@@ -107,13 +118,15 @@ func (*podSource) AddEventHandler(_ context.Context, _ func()) {
 }
 
 func (ps *podSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, error) {
-	pods, err := ps.podInformer.Lister().Pods(ps.namespace).List(labels.Everything())
-	if err != nil {
-		return nil, err
-	}
+	indexKeys := ps.podInformer.Informer().GetIndexer().ListIndexFuncValues(informers.IndexWithSelectors)
 
 	endpointMap := make(map[endpoint.EndpointKey][]string)
-	for _, pod := range pods {
+	for _, key := range indexKeys {
+		pod, err := informers.GetByKey[*corev1.Pod](ps.podInformer.Informer().GetIndexer(), key)
+		if err != nil {
+			continue
+		}
+
 		if ps.fqdnTemplate == nil || ps.combineFQDNAnnotation {
 			ps.addPodEndpointsToEndpointMap(endpointMap, pod)
 		}

source/pod_fqdn_test.go

@@ -58,7 +58,9 @@ func TestNewPodSourceWithFqdn(t *testing.T) {
 				false,
 				"",
 				tt.fqdnTemplate,
-				false)
+				false,
+				"",
+				nil)
 
 			if tt.expectError {
 				assert.Error(t, err)
@@ -405,7 +407,9 @@ func TestPodSourceFqdnTemplatingExamples(t *testing.T) {
 				false,
 				tt.sourceDomain,
 				tt.fqdnTemplate,
-				tt.combineFQDN)
+				tt.combineFQDN,
+				"",
+				nil)
 			require.NoError(t, err)
 
 			endpoints, err := src.Endpoints(t.Context())
@@ -467,7 +471,9 @@ func TestPodSourceFqdnTemplatingExamples_Failed(t *testing.T) {
 				false,
 				tt.sourceDomain,
 				tt.fqdnTemplate,
-				tt.combineFQDN)
+				tt.combineFQDN,
+				"",
+				nil)
 			require.NoError(t, err)
 
 			_, err = src.Endpoints(t.Context())

source/pod_indexer_test.go

@@ -0,0 +1,232 @@
+/*
+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 source
+
+import (
+	"fmt"
+	"math/rand/v2"
+	"net"
+	"strconv"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+	"sigs.k8s.io/external-dns/source/annotations"
+)
+
+type podSpec struct {
+	namespace   string
+	labels      map[string]string
+	annotations map[string]string
+	// with labels and annotations
+	totalTarget int
+	// without provided labels and annotations
+	totalRandom int
+}
+
+func fixtureCreatePodsWithNodes(input []podSpec) []*corev1.Pod {
+	var pods []*corev1.Pod
+
+	var createPod = func(index int, spec podSpec) *corev1.Pod {
+		return &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      fmt.Sprintf("pod-%d-%s", index, uuid.NewString()),
+				Namespace: spec.namespace,
+				Labels: func() map[string]string {
+					if spec.totalTarget > index {
+						return spec.labels
+					}
+					return map[string]string{
+						"app":   fmt.Sprintf("my-app-%d", rand.IntN(10)),
+						"index": strconv.Itoa(index),
+					}
+				}(),
+				Annotations: func() map[string]string {
+					if spec.totalTarget > index {
+						return spec.annotations
+					}
+					return map[string]string{
+						"key1": fmt.Sprintf("value-%d", rand.IntN(10)),
+					}
+				}(),
+			},
+			Spec: corev1.PodSpec{},
+			Status: corev1.PodStatus{
+				Phase: corev1.PodRunning,
+				PodIPs: []corev1.PodIP{
+					{IP: net.IPv4(192, byte(rand.IntN(250)), byte(rand.IntN(250)), byte(index)).String()},
+				},
+			},
+		}
+	}
+
+	for _, el := range input {
+		totalPods := el.totalTarget + el.totalRandom
+		for i := 0; i < totalPods; i++ {
+			pods = append(pods, createPod(i, el))
+		}
+	}
+
+	for i := 0; i < 3; i++ {
+		rand.Shuffle(len(pods), func(i, j int) {
+			pods[i], pods[j] = pods[j], pods[i]
+		})
+	}
+	// assign nodes to pods
+	for i, pod := range pods {
+		pod.Spec.NodeName = fmt.Sprintf("node-%d", i/5) // Assign 5 pods per node
+	}
+	return pods
+}
+
+func TestPodsWithAnnotationsAndLabels(t *testing.T) {
+	// total target pods 700
+	// total random pods 3950
+	pods := fixtureCreatePodsWithNodes([]podSpec{
+		{
+			namespace:   "dev",
+			labels:      map[string]string{"app": "nginx", "env": "dev", "agent": "enabled"},
+			annotations: map[string]string{"arch": "amd64"},
+			totalTarget: 300,
+			totalRandom: 700,
+		},
+		{
+			namespace:   "prod",
+			labels:      map[string]string{"app": "nginx", "env": "prod", "agent": "enabled"},
+			annotations: map[string]string{"arch": "amd64"},
+			totalTarget: 150,
+			totalRandom: 2700,
+		},
+		{
+			namespace:   "default",
+			labels:      map[string]string{"app": "nginx", "agent": "disabled"},
+			annotations: map[string]string{"arch": "amd64"},
+			totalTarget: 250,
+			totalRandom: 450,
+		},
+		{
+			namespace:   "kube-system",
+			labels:      map[string]string{},
+			annotations: map[string]string{},
+			totalTarget: 0,
+			totalRandom: 100,
+		},
+	})
+
+	client := fake.NewClientset()
+
+	nodes := map[string]bool{}
+
+	for _, pod := range pods {
+		if _, exists := nodes[pod.Spec.NodeName]; !exists {
+			nodes[pod.Spec.NodeName] = true
+			node := &corev1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: pod.Spec.NodeName,
+				},
+			}
+			if _, err := client.CoreV1().Nodes().Create(t.Context(), node, metav1.CreateOptions{}); err != nil {
+				assert.NoError(t, err)
+			}
+		}
+		if _, err := client.CoreV1().Pods(pod.Namespace).Create(t.Context(), pod, metav1.CreateOptions{}); err != nil {
+			assert.NoError(t, err)
+		}
+	}
+
+	tests := []struct {
+		name                  string
+		namespace             string
+		labelSelector         string
+		annotationFilter      string
+		expectedEndpointCount int
+	}{
+		{
+			name:                  "prod namespace with labels",
+			namespace:             "prod",
+			labelSelector:         "app=nginx",
+			expectedEndpointCount: 150,
+		},
+		{
+			name:                  "prod namespace with annotations",
+			namespace:             "prod",
+			annotationFilter:      "arch=amd64",
+			expectedEndpointCount: 150,
+		},
+		{
+			name:                  "prod namespace with annotations and labels not exists",
+			namespace:             "prod",
+			labelSelector:         "app=not-exists",
+			annotationFilter:      "arch=amd64",
+			expectedEndpointCount: 0,
+		},
+		{
+			name:                  "all namespaces with correct annotations and labels",
+			namespace:             "",
+			labelSelector:         "app=nginx,agent=enabled",
+			annotationFilter:      "arch=amd64",
+			expectedEndpointCount: 450, // 300 from dev + 150 from prod
+		},
+		{
+			name:                  "all namespaces with loose annotations and labels",
+			namespace:             "",
+			labelSelector:         "app=nginx",
+			annotationFilter:      "arch=amd64",
+			expectedEndpointCount: 700, // 300 from dev + 150 from prod + 250 from default
+		},
+		{
+			name:                  "all namespaces with loose annotations and labels",
+			namespace:             "",
+			labelSelector:         "agent",
+			annotationFilter:      "arch",
+			expectedEndpointCount: 700,
+		},
+		{
+			name:                  "all namespaces without filters",
+			namespace:             "",
+			labelSelector:         "",
+			annotationFilter:      "",
+			expectedEndpointCount: 4650,
+		},
+		{
+			name:                  "single namespace without filters",
+			namespace:             "default",
+			labelSelector:         "",
+			annotationFilter:      "",
+			expectedEndpointCount: 700,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			selector, _ := annotations.ParseFilter(tt.labelSelector)
+			pSource, err := NewPodSource(
+				t.Context(), client,
+				tt.namespace, "",
+				false, "",
+				"{{ .Name }}.tld.org", false,
+				tt.annotationFilter, selector)
+			require.NoError(t, err)
+
+			endpoints, err := pSource.Endpoints(t.Context())
+			require.NoError(t, err)
+
+			assert.Len(t, endpoints, tt.expectedEndpointCount)
+		})
+	}
+}

source/pod_test.go

@@ -657,7 +657,7 @@ func TestPodSource(t *testing.T) {
 				}
 			}
 
-			client, err := NewPodSource(ctx, kubernetes, tc.targetNamespace, tc.compatibility, tc.ignoreNonHostNetworkPods, tc.PodSourceDomain, "", false)
+			client, err := NewPodSource(ctx, kubernetes, tc.targetNamespace, tc.compatibility, tc.ignoreNonHostNetworkPods, tc.PodSourceDomain, "", false, "", nil)
 			require.NoError(t, err)
 
 			endpoints, err := client.Endpoints(ctx)
@@ -885,7 +885,7 @@ func TestPodSourceLogs(t *testing.T) {
 				}
 			}
 
-			client, err := NewPodSource(ctx, kubernetes, "", "", tc.ignoreNonHostNetworkPods, "", "", false)
+			client, err := NewPodSource(ctx, kubernetes, "", "", tc.ignoreNonHostNetworkPods, "", "", false, "", nil)
 			require.NoError(t, err)
 
 			hook := testutils.LogsUnderTestWithLogLevel(log.DebugLevel, t)

source/store.go

@@ -448,7 +448,7 @@ func buildPodSource(ctx context.Context, p ClientGenerator, cfg *Config) (Source
 	if err != nil {
 		return nil, err
 	}
-	return NewPodSource(ctx, client, cfg.Namespace, cfg.Compatibility, cfg.IgnoreNonHostNetworkPods, cfg.PodSourceDomain, cfg.FQDNTemplate, cfg.CombineFQDNAndAnnotation)
+	return NewPodSource(ctx, client, cfg.Namespace, cfg.Compatibility, cfg.IgnoreNonHostNetworkPods, cfg.PodSourceDomain, cfg.FQDNTemplate, cfg.CombineFQDNAndAnnotation, cfg.AnnotationFilter, cfg.LabelFilter)
 }
 
 // buildIstioGatewaySource creates an Istio Gateway source for exposing Istio gateways as DNS records.