chore(source/service): restructure code to make service type filters testable (#5485)

* chore(source/service): restructure code with filters are testable

* chore(source/service): restructure code with filters are testable

Co-authored-by: Michel Loiseleur <[email protected]>

* chore(source/service): restructure code with filters are testable

Signed-off-by: ivan katliarchuk <[email protected]>

---------

Signed-off-by: ivan katliarchuk <[email protected]>
Co-authored-by: Michel Loiseleur <[email protected]>

Author: ivankatliarchuk

docs/flags.md

@@ -46,7 +46,7 @@
 | `--pod-source-domain=""` | Domain to use for pods records (optional) |
 | `--[no-]publish-host-ip` | Allow external-dns to publish host-ip for headless services (optional) |
 | `--[no-]publish-internal-services` | Allow external-dns to publish DNS records for ClusterIP services (optional) |
-| `--service-type-filter=SERVICE-TYPE-FILTER` | The service types to take care about (default: all, expected: ClusterIP, NodePort, LoadBalancer or ExternalName) |
+| `--service-type-filter=SERVICE-TYPE-FILTER` | The service types to filter by. Specify multiple times for multiple filters to be applied. (optional, default: all, expected: ClusterIP, NodePort, LoadBalancer or ExternalName) |
 | `--source=source` | The resource types that are queried for endpoints; specify multiple times for multiple sources (required, options: service, ingress, node, pod, fake, connector, gateway-httproute, gateway-grpcroute, gateway-tlsroute, gateway-tcproute, gateway-udproute, istio-gateway, istio-virtualservice, cloudfoundry, contour-httpproxy, gloo-proxy, crd, empty, skipper-routegroup, openshift-route, ambassador-host, kong-tcpingress, f5-virtualserver, f5-transportserver, traefik-proxy) |
 | `--target-net-filter=TARGET-NET-FILTER` | Limit possible targets by a net filter; specify multiple times for multiple possible nets (optional) |
 | `--[no-]traefik-disable-legacy` | Disable listeners on Resources under the traefik.containo.us API Group |

pkg/apis/externaldns/types.go

@@ -480,7 +480,7 @@ func App(cfg *Config) *kingpin.Application {
 	app.Flag("pod-source-domain", "Domain to use for pods records (optional)").Default(defaultConfig.PodSourceDomain).StringVar(&cfg.PodSourceDomain)
 	app.Flag("publish-host-ip", "Allow external-dns to publish host-ip for headless services (optional)").BoolVar(&cfg.PublishHostIP)
 	app.Flag("publish-internal-services", "Allow external-dns to publish DNS records for ClusterIP services (optional)").BoolVar(&cfg.PublishInternal)
-	app.Flag("service-type-filter", "The service types to take care about (default: all, expected: ClusterIP, NodePort, LoadBalancer or ExternalName)").StringsVar(&cfg.ServiceTypeFilter)
+	app.Flag("service-type-filter", "The service types to filter by. Specify multiple times for multiple filters to be applied. (optional, default: all, expected: ClusterIP, NodePort, LoadBalancer or ExternalName)").Default(defaultConfig.ServiceTypeFilter...).StringsVar(&cfg.ServiceTypeFilter)
 	app.Flag("source", "The resource types that are queried for endpoints; specify multiple times for multiple sources (required, options: service, ingress, node, pod, fake, connector, gateway-httproute, gateway-grpcroute, gateway-tlsroute, gateway-tcproute, gateway-udproute, istio-gateway, istio-virtualservice, cloudfoundry, contour-httpproxy, gloo-proxy, crd, empty, skipper-routegroup, openshift-route, ambassador-host, kong-tcpingress, f5-virtualserver, f5-transportserver, traefik-proxy)").Required().PlaceHolder("source").EnumsVar(&cfg.Sources, "service", "ingress", "node", "pod", "gateway-httproute", "gateway-grpcroute", "gateway-tlsroute", "gateway-tcproute", "gateway-udproute", "istio-gateway", "istio-virtualservice", "cloudfoundry", "contour-httpproxy", "gloo-proxy", "fake", "connector", "crd", "empty", "skipper-routegroup", "openshift-route", "ambassador-host", "kong-tcpingress", "f5-virtualserver", "f5-transportserver", "traefik-proxy")
 	app.Flag("target-net-filter", "Limit possible targets by a net filter; specify multiple times for multiple possible nets (optional)").StringsVar(&cfg.TargetNetFilter)
 	app.Flag("traefik-disable-legacy", "Disable listeners on Resources under the traefik.containo.us API Group").Default(strconv.FormatBool(defaultConfig.TraefikDisableLegacy)).BoolVar(&cfg.TraefikDisableLegacy)

source/service.go

@@ -19,7 +19,9 @@ package source
 import (
 	"context"
 	"fmt"
+	"maps"
 	"net"
+	"slices"
 	"sort"
 	"strings"
 	"text/template"
@@ -41,20 +43,28 @@ import (
 	"sigs.k8s.io/external-dns/source/fqdn"
 )
 
+var (
+	knownServiceTypes = map[v1.ServiceType]struct{}{
+		v1.ServiceTypeClusterIP:    {}, // Default service type exposes the service on a cluster-internal IP.
+		v1.ServiceTypeNodePort:     {}, // Exposes the service on each node's IP at a static port.
+		v1.ServiceTypeLoadBalancer: {}, // Exposes the service externally using a cloud provider's load balancer.
+		v1.ServiceTypeExternalName: {}, // Maps the service to an external DNS name.
+	}
+)
+
 // serviceSource is an implementation of Source for Kubernetes service objects.
 // It will find all services that are under our jurisdiction, i.e. annotated
 // desired hostname and matching or no controller annotation. For each of the
 // matched services' entrypoints it will return a corresponding
 // Endpoint object.
 type serviceSource struct {
-	client           kubernetes.Interface
-	namespace        string
-	annotationFilter string
+	client                kubernetes.Interface
+	namespace             string
+	annotationFilter      string
+	labelSelector         labels.Selector
+	fqdnTemplate          *template.Template
+	combineFQDNAnnotation bool
 
-	// process Services with legacy annotations
-	compatibility                  string
-	fqdnTemplate                   *template.Template
-	combineFQDNAnnotation          bool
 	ignoreHostnameAnnotation       bool
 	publishInternal                bool
 	publishHostIP                  bool
@@ -65,9 +75,11 @@ type serviceSource struct {
 	endpointsInformer              coreinformers.EndpointsInformer
 	podInformer                    coreinformers.PodInformer
 	nodeInformer                   coreinformers.NodeInformer
-	serviceTypeFilter              map[string]struct{}
-	labelSelector                  labels.Selector
+	serviceTypeFilter              *serviceTypes
 	exposeInternalIPv6             bool
+
+	// process Services with legacy annotations
+	compatibility string
 }
 
 // NewServiceSource creates a new serviceSource with the given config.
@@ -78,7 +90,7 @@ func NewServiceSource(ctx context.Context, kubeClient kubernetes.Interface, name
 	}
 
 	// Use shared informers to listen for add/update/delete of services/pods/nodes in the specified namespace.
-	// Set resync period to 0, to prevent processing when nothing has changed
+	// Set the resync period to 0 to prevent processing when nothing has changed
 	informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, 0, kubeinformers.WithNamespace(namespace))
 	serviceInformer := informerFactory.Core().V1().Services()
 	endpointsInformer := informerFactory.Core().V1().Endpoints()
@@ -118,11 +130,10 @@ func NewServiceSource(ctx context.Context, kubeClient kubernetes.Interface, name
 		return nil, err
 	}
 
-	// Transform the slice into a map so it will
-	// be way much easier and fast to filter later
-	serviceTypes := make(map[string]struct{})
-	for _, serviceType := range serviceTypeFilter {
-		serviceTypes[serviceType] = struct{}{}
+	// Transform the slice into a map so it will be way much easier and fast to filter later
+	sTypesFilter, err := newServiceTypesFilter(serviceTypeFilter)
+	if err != nil {
+		return nil, err
 	}
 
 	return &serviceSource{
@@ -140,30 +151,29 @@ func NewServiceSource(ctx context.Context, kubeClient kubernetes.Interface, name
 		endpointsInformer:              endpointsInformer,
 		podInformer:                    podInformer,
 		nodeInformer:                   nodeInformer,
-		serviceTypeFilter:              serviceTypes,
+		serviceTypeFilter:              sTypesFilter,
 		labelSelector:                  labelSelector,
 		resolveLoadBalancerHostname:    resolveLoadBalancerHostname,
 		listenEndpointEvents:           listenEndpointEvents,
 		exposeInternalIPv6:             exposeInternalIPv6,
 	}, nil
 }
 
-// Endpoints returns endpoint objects for each service that should be processed.
-func (sc *serviceSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error) {
+// Endpoints return endpoint objects for each service that should be processed.
+func (sc *serviceSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, error) {
 	services, err := sc.serviceInformer.Lister().Services(sc.namespace).List(sc.labelSelector)
 	if err != nil {
 		return nil, err
 	}
+
+	// filter on service types if at least one has been provided
+	services = sc.filterByServiceType(services)
+
 	services, err = sc.filterByAnnotations(services)
 	if err != nil {
 		return nil, err
 	}
 
-	// filter on service types if at least one has been provided
-	if len(sc.serviceTypeFilter) > 0 {
-		services = sc.filterByServiceType(services)
-	}
-
 	endpoints := []*endpoint.Endpoint{}
 
 	for _, svc := range services {
@@ -427,11 +437,7 @@ func (sc *serviceSource) endpoints(svc *v1.Service) []*endpoint.Endpoint {
 
 // filterByAnnotations filters a list of services by a given annotation selector.
 func (sc *serviceSource) filterByAnnotations(services []*v1.Service) ([]*v1.Service, error) {
-	labelSelector, err := metav1.ParseToLabelSelector(sc.annotationFilter)
-	if err != nil {
-		return nil, err
-	}
-	selector, err := metav1.LabelSelectorAsSelector(labelSelector)
+	selector, err := annotations.ParseFilter(sc.annotationFilter)
 	if err != nil {
 		return nil, err
 	}
@@ -449,21 +455,23 @@ func (sc *serviceSource) filterByAnnotations(services []*v1.Service) ([]*v1.Serv
 			filteredList = append(filteredList, service)
 		}
 	}
-
+	log.Debugf("filtered %d services out of %d with annotation filter", len(filteredList), len(services))
 	return filteredList, nil
 }
 
-// filterByServiceType filters services according their types
+// filterByServiceType filters services according to their types
 func (sc *serviceSource) filterByServiceType(services []*v1.Service) []*v1.Service {
-	filteredList := []*v1.Service{}
+	if !sc.serviceTypeFilter.enabled || len(services) == 0 {
+		return services
+	}
+	var result []*v1.Service
 	for _, service := range services {
-		// Check if the service is of the given type or not
-		if _, ok := sc.serviceTypeFilter[string(service.Spec.Type)]; ok {
-			filteredList = append(filteredList, service)
+		if _, ok := sc.serviceTypeFilter.types[service.Spec.Type]; ok {
+			result = append(result, service)
 		}
 	}
-
-	return filteredList
+	log.Debugf("filtered %d services out of %d with service types filter %q", len(result), len(services), slices.Collect(maps.Keys(sc.serviceTypeFilter.types)))
+	return result
 }
 
 func (sc *serviceSource) setResourceLabel(service *v1.Service, endpoints []*endpoint.Endpoint) {
@@ -625,9 +633,9 @@ func (sc *serviceSource) nodesExternalTrafficPolicyTypeLocal(svc *v1.Service) []
 	// If none available, fall back to nodes with ready pods
 	// If still none, use nodes with any running pods
 	if len(nodes) > 0 {
-		// Works same as service endpoints
+		// Works the same as service endpoints
 	} else if len(nodesReady) > 0 {
-		// 2 level of panic modes as safe guard, because old wrong behavior can be used by someone
+		// 2 level of panic modes as safeguard, because old wrong behavior can be used by someone
 		// Publish all endpoints not always a bad thing
 		log.Debugf("All pods in terminating state, use ready")
 		nodes = nodesReady
@@ -639,7 +647,7 @@ func (sc *serviceSource) nodesExternalTrafficPolicyTypeLocal(svc *v1.Service) []
 	return nodes
 }
 
-// pods retrieves a slice of pods associated with the given Service
+// pods retrieve a slice of pods associated with the given Service
 func (sc *serviceSource) pods(svc *v1.Service) []*v1.Pod {
 	labelSelector, err := metav1.ParseToLabelSelector(labels.Set(svc.Spec.Selector).AsSelectorPreValidated().String())
 	if err != nil {
@@ -755,3 +763,31 @@ func (sc *serviceSource) AddEventHandler(_ context.Context, handler func()) {
 		sc.endpointsInformer.Informer().AddEventHandler(eventHandlerFunc(handler))
 	}
 }
+
+type serviceTypes struct {
+	enabled bool
+	types   map[v1.ServiceType]bool
+}
+
+// newServiceTypesFilter processes a slice of service type filter strings and returns a serviceTypes struct.
+// It validates the filter against known Kubernetes service types. If the filter is empty or contains an empty string,
+// service type filtering is disabled. If an unknown type is found, an error is returned.
+func newServiceTypesFilter(filter []string) (*serviceTypes, error) {
+	if len(filter) == 0 || slices.Contains(filter, "") {
+		return &serviceTypes{
+			enabled: false,
+		}, nil
+	}
+	types := make(map[v1.ServiceType]bool)
+	for _, serviceType := range filter {
+		if _, ok := knownServiceTypes[v1.ServiceType(serviceType)]; !ok {
+			return nil, fmt.Errorf("unsupported service type filter: %q. Supported types are: %q", serviceType, slices.Collect(maps.Keys(knownServiceTypes)))
+		}
+		types[v1.ServiceType(serviceType)] = true
+	}
+
+	return &serviceTypes{
+		enabled: true,
+		types:   types,
+	}, nil
+}