15617390418

Committed 12 Jun 2025 05:38PM UTC coverage: 76.259% (+0.09%) from 76.172%

Build # 15617390418

Build Type

Pull #5507

github

Committed by

ivankatliarchuk

Commit Message

chore(source/ingress): add fqdn specific tests for ingress source

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

Pull Request Pull Request #5507: chore(source/ingress): add fqdn specific tests for ingress source

Run Details

7 of 7 new or added lines in 2 files covered. (100.0%)

41 existing lines in 5 files now uncovered.

14162 of 18571 relevant lines covered (76.26%)

759.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.49

/source/node.go

/*
Copyright 2017 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 (
        "context"
        "fmt"
        "text/template"

        log "github.com/sirupsen/logrus"
        v1 "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/endpoint"
        "sigs.k8s.io/external-dns/source/annotations"
        "sigs.k8s.io/external-dns/source/fqdn"
        "sigs.k8s.io/external-dns/source/informers"
)

const warningMsg = "The default behavior of exposing internal IPv6 addresses will change in the next minor version. Use --no-expose-internal-ipv6 flag to opt-in to the new behavior."

type nodeSource struct {
        client               kubernetes.Interface
        annotationFilter     string
        fqdnTemplate         *template.Template
        nodeInformer         coreinformers.NodeInformer
        labelSelector        labels.Selector
        excludeUnschedulable bool
        exposeInternalIPv6   bool
}

// NewNodeSource creates a new nodeSource with the given config.
func NewNodeSource(ctx context.Context, kubeClient kubernetes.Interface, annotationFilter, fqdnTemplate string, labelSelector labels.Selector, exposeInternalIPv6 bool, excludeUnschedulable bool) (Source, error) {
        tmpl, err := fqdn.ParseTemplate(fqdnTemplate)
        if err != nil {
                return nil, err
        }

        // Use shared informers to listen for add/update/delete of nodes.
        // Set resync period to 0, to prevent processing when nothing has changed
        informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, 0)
        nodeInformer := informerFactory.Core().V1().Nodes()

        // Add default resource event handler to properly initialize informer.
        nodeInformer.Informer().AddEventHandler(
                cache.ResourceEventHandlerFuncs{
                        AddFunc: func(obj interface{}) {
                                log.Debug("node added")
                        },
                },
        )

        informerFactory.Start(ctx.Done())

        // wait for the local cache to be populated.
        if err := informers.WaitForCacheSync(context.Background(), informerFactory); err != nil {
                return nil, err
        }

        return &nodeSource{
                client:               kubeClient,
                annotationFilter:     annotationFilter,
                fqdnTemplate:         tmpl,
                nodeInformer:         nodeInformer,
                labelSelector:        labelSelector,
                excludeUnschedulable: excludeUnschedulable,
                exposeInternalIPv6:   exposeInternalIPv6,
        }, nil
}

// Endpoints returns endpoint objects for each service that should be processed.
func (ns *nodeSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error) {
        nodes, err := ns.nodeInformer.Lister().List(ns.labelSelector)
        if err != nil {
                return nil, err
        }

        nodes, err = ns.filterByAnnotations(nodes)
        if err != nil {
                return nil, err
        }

        endpoints := map[endpoint.EndpointKey]*endpoint.Endpoint{}

        // create endpoints for all nodes
        for _, node := range nodes {
                // Check controller annotation to see if we are responsible.
                controller, ok := node.Annotations[controllerAnnotationKey]
                if ok && controller != controllerAnnotationValue {
                        log.Debugf("Skipping node %s because controller value does not match, found: %s, required: %s",
                                node.Name, controller, controllerAnnotationValue)
                        continue
                }

                if node.Spec.Unschedulable && ns.excludeUnschedulable {
                        log.Debugf("Skipping node %s because it is unschedulable", node.Name)
                        continue
                }

                log.Debugf("creating endpoint for node %s", node.Name)

                ttl := annotations.TTLFromAnnotations(node.Annotations, fmt.Sprintf("node/%s", node.Name))

                // create new endpoint with the information we already have
                ep := &endpoint.Endpoint{
                        RecordTTL: ttl,
                }

                if ns.fqdnTemplate != nil {
                        hostnames, err := fqdn.ExecTemplate(ns.fqdnTemplate, node)
                        if err != nil {
                                return nil, err
                        }
                        hostname := ""
                        if len(hostnames) > 0 {
                                hostname = hostnames[0]
                        }
                        ep.DNSName = hostname
                        log.Debugf("applied template for %s, converting to %s", node.Name, ep.DNSName)
                } else {
                        ep.DNSName = node.Name
                        log.Debugf("not applying template for %s", node.Name)
                }

                addrs := annotations.TargetsFromTargetAnnotation(node.Annotations)
                if len(addrs) == 0 {
                        addrs, err = ns.nodeAddresses(node)
                        if err != nil {
                                return nil, fmt.Errorf("failed to get node address from %s: %w", node.Name, err)
                        }
                }

                ep.Labels = endpoint.NewLabels()
                for _, addr := range addrs {
                        log.Debugf("adding endpoint %s target %s", ep, addr)
                        key := endpoint.EndpointKey{
                                DNSName:    ep.DNSName,
                                RecordType: suitableType(addr),
                        }
                        if _, ok := endpoints[key]; !ok {
                                epCopy := *ep
                                epCopy.RecordType = key.RecordType
                                endpoints[key] = &epCopy
                        }
                        endpoints[key].Targets = append(endpoints[key].Targets, addr)
                }
        }

        endpointsSlice := []*endpoint.Endpoint{}
        for _, ep := range endpoints {
                endpointsSlice = append(endpointsSlice, ep)
        }

        return endpointsSlice, nil
}

func (ns *nodeSource) AddEventHandler(ctx context.Context, handler func()) {
}

// nodeAddress returns node's externalIP and if that's not found, node's internalIP
// basically what k8s.io/kubernetes/pkg/util/node.GetPreferredNodeAddress does
func (ns *nodeSource) nodeAddresses(node *v1.Node) ([]string, error) {
        addresses := map[v1.NodeAddressType][]string{
                v1.NodeExternalIP: {},
                v1.NodeInternalIP: {},
        }
        var internalIpv6Addresses []string

        for _, addr := range node.Status.Addresses {
                // IPv6 InternalIP addresses have special handling.
                // Refer to https://github.com/kubernetes-sigs/external-dns/pull/5192 for more details.
                if addr.Type == v1.NodeInternalIP && suitableType(addr.Address) == endpoint.RecordTypeAAAA {
                        internalIpv6Addresses = append(internalIpv6Addresses, addr.Address)
                }
                addresses[addr.Type] = append(addresses[addr.Type], addr.Address)
        }

        if len(addresses[v1.NodeExternalIP]) > 0 {
                if ns.exposeInternalIPv6 {
                        log.Warn(warningMsg)
                        return append(addresses[v1.NodeExternalIP], internalIpv6Addresses...), nil
                }
                return addresses[v1.NodeExternalIP], nil
        }

        if len(addresses[v1.NodeInternalIP]) > 0 {
                return addresses[v1.NodeInternalIP], nil
        }

        return nil, fmt.Errorf("could not find node address for %s", node.Name)
}

// filterByAnnotations filters a list of nodes by a given annotation selector.
func (ns *nodeSource) filterByAnnotations(nodes []*v1.Node) ([]*v1.Node, error) {
        selector, err := annotations.ParseFilter(ns.annotationFilter)
        if err != nil {
                return nil, err
        }

        // empty filter returns original list
        if selector.Empty() {
                return nodes, nil
        }

        var filteredList []*v1.Node

        for _, node := range nodes {
                // include node if its annotations match the selector
                if selector.Matches(labels.Set(node.Annotations)) {
                        filteredList = append(filteredList, node)
                }
        }

        return filteredList, nil
}

1	/*
2	Copyright 2017 The Kubernetes Authors.
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package source
18
19	import (
20	"context"
21	"fmt"
22	"text/template"
23
24	log "github.com/sirupsen/logrus"
25	v1 "k8s.io/api/core/v1"
26	"k8s.io/apimachinery/pkg/labels"
27	kubeinformers "k8s.io/client-go/informers"
28	coreinformers "k8s.io/client-go/informers/core/v1"
29	"k8s.io/client-go/kubernetes"
30	"k8s.io/client-go/tools/cache"
31
32	"sigs.k8s.io/external-dns/endpoint"
33	"sigs.k8s.io/external-dns/source/annotations"
34	"sigs.k8s.io/external-dns/source/fqdn"
35	"sigs.k8s.io/external-dns/source/informers"
36	)
37
38	const warningMsg = "The default behavior of exposing internal IPv6 addresses will change in the next minor version. Use --no-expose-internal-ipv6 flag to opt-in to the new behavior."
39
40	type nodeSource struct {
41	client kubernetes.Interface
42	annotationFilter string
43	fqdnTemplate *template.Template
44	nodeInformer coreinformers.NodeInformer
45	labelSelector labels.Selector
46	excludeUnschedulable bool
47	exposeInternalIPv6 bool
48	}
49
50	// NewNodeSource creates a new nodeSource with the given config.
51	func NewNodeSource(ctx context.Context, kubeClient kubernetes.Interface, annotationFilter, fqdnTemplate string, labelSelector labels.Selector, exposeInternalIPv6 bool, excludeUnschedulable bool) (Source, error) {	48✔
52	tmpl, err := fqdn.ParseTemplate(fqdnTemplate)	48✔
53	if err != nil {	50✔
54	return nil, err	2✔
55	}	2✔
56
57	// Use shared informers to listen for add/update/delete of nodes.
58	// Set resync period to 0, to prevent processing when nothing has changed
59	informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, 0)	46✔
60	nodeInformer := informerFactory.Core().V1().Nodes()	46✔
61		46✔
62	// Add default resource event handler to properly initialize informer.	46✔
63	nodeInformer.Informer().AddEventHandler(	46✔
64	cache.ResourceEventHandlerFuncs{	46✔
65	AddFunc: func(obj interface{}) {	89✔
66	log.Debug("node added")	43✔
67	},	43✔
68	},
69	)
70
71	informerFactory.Start(ctx.Done())	46✔
72		46✔
73	// wait for the local cache to be populated.	46✔
74	if err := informers.WaitForCacheSync(context.Background(), informerFactory); err != nil {	46✔
75	return nil, err	×
76	}	×
77
78	return &nodeSource{	46✔
79	client: kubeClient,	46✔
80	annotationFilter: annotationFilter,	46✔
81	fqdnTemplate: tmpl,	46✔
82	nodeInformer: nodeInformer,	46✔
83	labelSelector: labelSelector,	46✔
84	excludeUnschedulable: excludeUnschedulable,	46✔
85	exposeInternalIPv6: exposeInternalIPv6,	46✔
86	}, nil	46✔
87	}
88
89	// Endpoints returns endpoint objects for each service that should be processed.
90	func (ns nodeSource) Endpoints(ctx context.Context) ([]endpoint.Endpoint, error) {	39✔
91	nodes, err := ns.nodeInformer.Lister().List(ns.labelSelector)	39✔
92	if err != nil {	39✔
93	return nil, err	×
94	}	×
95
96	nodes, err = ns.filterByAnnotations(nodes)	39✔
97	if err != nil {	39✔
98	return nil, err	×
99	}	×
100
101	endpoints := map[endpoint.EndpointKey]*endpoint.Endpoint{}	39✔
102		39✔
103	// create endpoints for all nodes	39✔
104	for _, node := range nodes {	80✔
105	// Check controller annotation to see if we are responsible.	41✔
106	controller, ok := node.Annotations[controllerAnnotationKey]	42✔
107	if ok && controller != controllerAnnotationValue {	1✔
108	log.Debugf("Skipping node %s because controller value does not match, found: %s, required: %s",	1✔
109	node.Name, controller, controllerAnnotationValue)	1✔
110	continue
111	}
112		41✔
113	if node.Spec.Unschedulable && ns.excludeUnschedulable {	1✔
114	log.Debugf("Skipping node %s because it is unschedulable", node.Name)	1✔
115	continue
116	}
117		39✔
118	log.Debugf("creating endpoint for node %s", node.Name)	39✔
119		39✔
120	ttl := annotations.TTLFromAnnotations(node.Annotations, fmt.Sprintf("node/%s", node.Name))	39✔
121		39✔
122	// create new endpoint with the information we already have	39✔
123	ep := &endpoint.Endpoint{	76✔
124	RecordTTL: ttl,	37✔
125	}	38✔
126		1✔
127	if ns.fqdnTemplate != nil {	1✔
128	hostnames, err := fqdn.ExecTemplate(ns.fqdnTemplate, node)
129	if err != nil {
130	return nil, err	38✔
131	}	38✔
132	hostname := ""	52✔
133	if len(hostnames) > 0 {	14✔
134	hostname = hostnames[0]	14✔
UNCOV 135	}	×
UNCOV 136	ep.DNSName = hostname	×
137	log.Debugf("applied template for %s, converting to %s", node.Name, ep.DNSName)
138	} else {	32✔
139	ep.DNSName = node.Name	18✔
140	log.Debugf("not applying template for %s", node.Name)	18✔
141	}	18✔
142		24✔
143	addrs := annotations.TargetsFromTargetAnnotation(node.Annotations)	24✔
144	if len(addrs) == 0 {	24✔
145	addrs, err = ns.nodeAddresses(node)	24✔
146	if err != nil {
147	return nil, fmt.Errorf("failed to get node address from %s: %w", node.Name, err)	80✔
148	}	42✔
149	}	42✔
150		101✔
151	ep.Labels = endpoint.NewLabels()	59✔
152	for _, addr := range addrs {	59✔
153	log.Debugf("adding endpoint %s target %s", ep, addr)	59✔
154	key := endpoint.EndpointKey{	59✔
155	DNSName: ep.DNSName,	59✔
156	RecordType: suitableType(addr),	59✔
157	}	59✔
158	if _, ok := endpoints[key]; !ok {	59✔
159	epCopy := *ep	116✔
160	epCopy.RecordType = key.RecordType	57✔
161	endpoints[key] = &epCopy	57✔
162	}	57✔
163	endpoints[key].Targets = append(endpoints[key].Targets, addr)	57✔
164	}	59✔
165	}
166
167	endpointsSlice := []*endpoint.Endpoint{}
168	for _, ep := range endpoints {
169	endpointsSlice = append(endpointsSlice, ep)	38✔
170	}	95✔
171		57✔
172	return endpointsSlice, nil	57✔
173	}
174		38✔
175	func (ns *nodeSource) AddEventHandler(ctx context.Context, handler func()) {
176	}
UNCOV 177		×
UNCOV 178	// nodeAddress returns node's externalIP and if that's not found, node's internalIP	×
179	// basically what k8s.io/kubernetes/pkg/util/node.GetPreferredNodeAddress does
180	func (ns nodeSource) nodeAddresses(node v1.Node) ([]string, error) {
181	addresses := map[v1.NodeAddressType][]string{
182	v1.NodeExternalIP: {},	37✔
183	v1.NodeInternalIP: {},	37✔
184	}	37✔
185	var internalIpv6Addresses []string	37✔
186		37✔
187	for _, addr := range node.Status.Addresses {	37✔
188	// IPv6 InternalIP addresses have special handling.	37✔
189	// Refer to https://github.com/kubernetes-sigs/external-dns/pull/5192 for more details.	92✔
190	if addr.Type == v1.NodeInternalIP && suitableType(addr.Address) == endpoint.RecordTypeAAAA {	55✔
191	internalIpv6Addresses = append(internalIpv6Addresses, addr.Address)	55✔
192	}	71✔
193	addresses[addr.Type] = append(addresses[addr.Type], addr.Address)	16✔
194	}	16✔
195		55✔
196	if len(addresses[v1.NodeExternalIP]) > 0 {
197	if ns.exposeInternalIPv6 {
198	log.Warn(warningMsg)	62✔
199	return append(addresses[v1.NodeExternalIP], internalIpv6Addresses...), nil	48✔
200	}	23✔
201	return addresses[v1.NodeExternalIP], nil	23✔
202	}	23✔
203		2✔
204	if len(addresses[v1.NodeInternalIP]) > 0 {
205	return addresses[v1.NodeInternalIP], nil
206	}	23✔
207		11✔
208	return nil, fmt.Errorf("could not find node address for %s", node.Name)	11✔
209	}
210		1✔
211	// filterByAnnotations filters a list of nodes by a given annotation selector.
212	func (ns nodeSource) filterByAnnotations(nodes []v1.Node) ([]*v1.Node, error) {
213	selector, err := annotations.ParseFilter(ns.annotationFilter)
214	if err != nil {	39✔
215	return nil, err	39✔
216	}	39✔
UNCOV 217		×
UNCOV 218	// empty filter returns original list	×
219	if selector.Empty() {
220	return nodes, nil
221	}	76✔
222		37✔
223	var filteredList []*v1.Node	37✔
224
225	for _, node := range nodes {	2✔
226	// include node if its annotations match the selector	2✔
227	if selector.Matches(labels.Set(node.Annotations)) {	4✔
228	filteredList = append(filteredList, node)	2✔
229	}	3✔
230	}	1✔
231		1✔
232	return filteredList, nil
233	}

kubernetes-sigs / external-dns / 15617390418

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous