15668101474

Committed 15 Jun 2025 10:36PM UTC coverage: 76.642% (+0.01%) from 76.63%

Build # 15668101474

Build Type

push

github

Committed by

web-flow

Commit Message

feat(source/node): fqdn support combineFQDNAnnotation (#5526)

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

Run Details

29 of 33 new or added lines in 2 files covered. (87.88%)

1 existing line in 1 file now uncovered.

14332 of 18700 relevant lines covered (76.64%)

759.52 hits per line

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

90.67

/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
        combineFQDNAnnotation bool

        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,
        excludeUnschedulable bool,
        combineFQDNAnnotation 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,
                combineFQDNAnnotation: combineFQDNAnnotation,
                nodeInformer:          nodeInformer,
                labelSelector:         labelSelector,
                excludeUnschedulable:  excludeUnschedulable,
                exposeInternalIPv6:    exposeInternalIPv6,
        }, nil
}

// Endpoints returns endpoint objects for each service that should be processed.
func (ns *nodeSource) Endpoints(_ 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 the controller annotation to see if we are responsible.
                if controller, ok := node.Annotations[controllerAnnotationKey]; 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))

                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)
                        }
                }

                dnsNames, err := ns.collectDNSNames(node)
                if err != nil {
                        return nil, err
                }

                for dns := range dnsNames {
                        log.Debugf("adding endpoint with %d targets", len(addrs))

                        for _, addr := range addrs {
                                ep := endpoint.NewEndpointWithTTL(dns, suitableType(addr), ttl)
                                ep.WithLabel(endpoint.ResourceLabelKey, fmt.Sprintf("node/%s", node.Name))

                                log.Debugf("adding endpoint %s target %s", ep, addr)
                                key := endpoint.EndpointKey{
                                        DNSName:    ep.DNSName,
                                        RecordType: ep.RecordType,
                                }
                                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(_ context.Context, _ func()) {
}

// nodeAddress returns the node's externalIP and if that's not found, the 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 a node if its annotations match the selector
                if selector.Matches(labels.Set(node.Annotations)) {
                        filteredList = append(filteredList, node)
                }
        }

        return filteredList, nil
}

// collectDNSNames returns a set of DNS names associated with the given Kubernetes Node.
// If an FQDN template is configured, it renders the template using the Node object
// to generate one or more DNS names.
// If combineFQDNAnnotation is enabled, the Node's name is also included alongside
// the templated names. If no FQDN template is provided, the result will include only
// the Node's name.
//
// Returns an error if template rendering fails.
func (ns *nodeSource) collectDNSNames(node *v1.Node) (map[string]bool, error) {
        dnsNames := make(map[string]bool)
        // If no FQDN template is configured, fallback to the node name
        if ns.fqdnTemplate == nil {
                dnsNames[node.Name] = true
                return dnsNames, nil
        }

        names, err := fqdn.ExecTemplate(ns.fqdnTemplate, node)
        if err != nil {
                return nil, err
        }

        for _, name := range names {
                dnsNames[name] = true
                log.Debugf("applied template for %s, converting to %s", node.Name, name)
        }

        if ns.combineFQDNAnnotation {
                dnsNames[node.Name] = true
        }

        return dnsNames, 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	combineFQDNAnnotation bool
45
46	nodeInformer coreinformers.NodeInformer
47	labelSelector labels.Selector
48	excludeUnschedulable bool
49	exposeInternalIPv6 bool
50	}
51
52	// NewNodeSource creates a new nodeSource with the given config.
53	func NewNodeSource(
54	ctx context.Context,
55	kubeClient kubernetes.Interface,
56	annotationFilter, fqdnTemplate string,
57	labelSelector labels.Selector,
58	exposeInternalIPv6,
59	excludeUnschedulable bool,
60	combineFQDNAnnotation bool) (Source, error) {	49✔
61	tmpl, err := fqdn.ParseTemplate(fqdnTemplate)	49✔
62	if err != nil {	51✔
63	return nil, err	2✔
64	}	2✔
65
66	// Use shared informers to listen for add/update/delete of nodes.
67	// Set resync period to 0, to prevent processing when nothing has changed
68	informerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, 0)	47✔
69	nodeInformer := informerFactory.Core().V1().Nodes()	47✔
70		47✔
71	// Add default resource event handler to properly initialize informer.	47✔
72	nodeInformer.Informer().AddEventHandler(	47✔
73	cache.ResourceEventHandlerFuncs{	47✔
74	AddFunc: func(obj interface{}) {	92✔
75	log.Debug("node added")	45✔
76	},	45✔
77	},
78	)
79
80	informerFactory.Start(ctx.Done())	47✔
81		47✔
82	// wait for the local cache to be populated.	47✔
83	if err := informers.WaitForCacheSync(context.Background(), informerFactory); err != nil {	47✔
84	return nil, err	×
85	}	×
86
87	return &nodeSource{	47✔
88	client: kubeClient,	47✔
89	annotationFilter: annotationFilter,	47✔
90	fqdnTemplate: tmpl,	47✔
91	combineFQDNAnnotation: combineFQDNAnnotation,	47✔
92	nodeInformer: nodeInformer,	47✔
93	labelSelector: labelSelector,	47✔
94	excludeUnschedulable: excludeUnschedulable,	47✔
95	exposeInternalIPv6: exposeInternalIPv6,	47✔
96	}, nil	47✔
97	}
98
99	// Endpoints returns endpoint objects for each service that should be processed.
100	func (ns nodeSource) Endpoints(_ context.Context) ([]endpoint.Endpoint, error) {	40✔
101	nodes, err := ns.nodeInformer.Lister().List(ns.labelSelector)	40✔
102	if err != nil {	40✔
103	return nil, err	×
104	}	×
105
106	nodes, err = ns.filterByAnnotations(nodes)	40✔
107	if err != nil {	40✔
108	return nil, err	×
109	}	×
110
111	endpoints := map[endpoint.EndpointKey]*endpoint.Endpoint{}	40✔
112		40✔
113	// create endpoints for all nodes	40✔
114	for _, node := range nodes {	83✔
115	// Check the controller annotation to see if we are responsible.	43✔
116	if controller, ok := node.Annotations[controllerAnnotationKey]; ok && controller != controllerAnnotationValue {	44✔
117	log.Debugf("Skipping node %s because controller value does not match, found: %s, required: %s",	1✔
118	node.Name, controller, controllerAnnotationValue)	1✔
119	continue	1✔
120	}
121
122	if node.Spec.Unschedulable && ns.excludeUnschedulable {	43✔
123	log.Debugf("Skipping node %s because it is unschedulable", node.Name)	1✔
124	continue	1✔
125	}
126
127	log.Debugf("creating endpoint for node %s", node.Name)	41✔
128		41✔
129	ttl := annotations.TTLFromAnnotations(node.Annotations, fmt.Sprintf("node/%s", node.Name))	41✔
130		41✔
131	addrs := annotations.TargetsFromTargetAnnotation(node.Annotations)	41✔
132		41✔
133	if len(addrs) == 0 {	80✔
134	addrs, err = ns.nodeAddresses(node)	39✔
135	if err != nil {	40✔
136	return nil, fmt.Errorf("failed to get node address from %s: %w", node.Name, err)	1✔
137	}	1✔
138	}
139
140	dnsNames, err := ns.collectDNSNames(node)	40✔
141	if err != nil {	40✔
NEW 142	return nil, err	×
UNCOV 143	}	×
144
145	for dns := range dnsNames {	88✔
146	log.Debugf("adding endpoint with %d targets", len(addrs))	48✔
147		48✔
148	for _, addr := range addrs {	113✔
149	ep := endpoint.NewEndpointWithTTL(dns, suitableType(addr), ttl)	65✔
150	ep.WithLabel(endpoint.ResourceLabelKey, fmt.Sprintf("node/%s", node.Name))	65✔
151		65✔
152	log.Debugf("adding endpoint %s target %s", ep, addr)	65✔
153	key := endpoint.EndpointKey{	65✔
154	DNSName: ep.DNSName,	65✔
155	RecordType: ep.RecordType,	65✔
156	}	65✔
157	if _, ok := endpoints[key]; !ok {	127✔
158	epCopy := *ep	62✔
159	epCopy.RecordType = key.RecordType	62✔
160	endpoints[key] = &epCopy	62✔
161	}	62✔
162	endpoints[key].Targets = append(endpoints[key].Targets, addr)	65✔
163	}
164	}
165	}
166
167	endpointsSlice := []*endpoint.Endpoint{}	39✔
168	for _, ep := range endpoints {	101✔
169	endpointsSlice = append(endpointsSlice, ep)	62✔
170	}	62✔
171
172	return endpointsSlice, nil	39✔
173	}
174
175	func (ns *nodeSource) AddEventHandler(_ context.Context, _ func()) {	×
176	}	×
177
178	// nodeAddress returns the node's externalIP and if that's not found, the node's internalIP
179	// basically what k8s.io/kubernetes/pkg/util/node.GetPreferredNodeAddress does
180	func (ns nodeSource) nodeAddresses(node v1.Node) ([]string, error) {	39✔
181	addresses := map[v1.NodeAddressType][]string{	39✔
182	v1.NodeExternalIP: {},	39✔
183	v1.NodeInternalIP: {},	39✔
184	}	39✔
185	var internalIpv6Addresses []string	39✔
186		39✔
187	for _, addr := range node.Status.Addresses {	96✔
188	// IPv6 InternalIP addresses have special handling.	57✔
189	// Refer to https://github.com/kubernetes-sigs/external-dns/pull/5192 for more details.	57✔
190	if addr.Type == v1.NodeInternalIP && suitableType(addr.Address) == endpoint.RecordTypeAAAA {	73✔
191	internalIpv6Addresses = append(internalIpv6Addresses, addr.Address)	16✔
192	}	16✔
193	addresses[addr.Type] = append(addresses[addr.Type], addr.Address)	57✔
194	}
195
196	if len(addresses[v1.NodeExternalIP]) > 0 {	66✔
197	if ns.exposeInternalIPv6 {	52✔
198	log.Warn(warningMsg)	25✔
199	return append(addresses[v1.NodeExternalIP], internalIpv6Addresses...), nil	25✔
200	}	25✔
201	return addresses[v1.NodeExternalIP], nil	2✔
202	}
203
204	if len(addresses[v1.NodeInternalIP]) > 0 {	23✔
205	return addresses[v1.NodeInternalIP], nil	11✔
206	}	11✔
207
208	return nil, fmt.Errorf("could not find node address for %s", node.Name)	1✔
209	}
210
211	// filterByAnnotations filters a list of nodes by a given annotation selector.
212	func (ns nodeSource) filterByAnnotations(nodes []v1.Node) ([]*v1.Node, error) {	40✔
213	selector, err := annotations.ParseFilter(ns.annotationFilter)	40✔
214	if err != nil {	40✔
215	return nil, err	×
216	}	×
217
218	// empty filter returns original list
219	if selector.Empty() {	78✔
220	return nodes, nil	38✔
221	}	38✔
222
223	var filteredList []*v1.Node	2✔
224		2✔
225	for _, node := range nodes {	4✔
226	// include a node if its annotations match the selector	2✔
227	if selector.Matches(labels.Set(node.Annotations)) {	3✔
228	filteredList = append(filteredList, node)	1✔
229	}	1✔
230	}
231
232	return filteredList, nil	2✔
233	}
234
235	// collectDNSNames returns a set of DNS names associated with the given Kubernetes Node.
236	// If an FQDN template is configured, it renders the template using the Node object
237	// to generate one or more DNS names.
238	// If combineFQDNAnnotation is enabled, the Node's name is also included alongside
239	// the templated names. If no FQDN template is provided, the result will include only
240	// the Node's name.
241	//
242	// Returns an error if template rendering fails.
243	func (ns nodeSource) collectDNSNames(node v1.Node) (map[string]bool, error) {	40✔
244	dnsNames := make(map[string]bool)	40✔
245	// If no FQDN template is configured, fallback to the node name	40✔
246	if ns.fqdnTemplate == nil {	64✔
247	dnsNames[node.Name] = true	24✔
248	return dnsNames, nil	24✔
249	}	24✔
250
251	names, err := fqdn.ExecTemplate(ns.fqdnTemplate, node)	16✔
252	if err != nil {	16✔
NEW 253	return nil, err	×
NEW 254	}	×
255
256	for _, name := range names {	38✔
257	dnsNames[name] = true	22✔
258	log.Debugf("applied template for %s, converting to %s", node.Name, name)	22✔
259	}	22✔
260
261	if ns.combineFQDNAnnotation {	18✔
262	dnsNames[node.Name] = true	2✔
263	}	2✔
264
265	return dnsNames, nil	16✔
266	}

kubernetes-sigs / external-dns / 15668101474

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