15497247364

Committed 06 Jun 2025 06:28PM UTC coverage: 76.242% (+0.003%) from 76.239%

Build # 15497247364

Build Type

push

github

Committed by

web-flow

Commit Message

chore(source/node): template expansion (#5498)

Coverage Stats

32 of 33 new or added lines in 1 file covered. (96.97%)

14146 of 18554 relevant lines covered (76.24%)

760.21 hits per line

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

91.37

/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, 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.
                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 := make(map[string]bool)

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

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

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

                        for _, addr := range addrs {
                                ep := endpoint.NewEndpointWithTTL(dns, suitableType(addr), ttl)
                                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
}

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, 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	if controller, ok := node.Annotations[controllerAnnotationKey]; ok && controller != controllerAnnotationValue {	42✔
107	log.Debugf("Skipping node %s because controller value does not match, found: %s, required: %s",	1✔
108	node.Name, controller, controllerAnnotationValue)	1✔
109	continue	1✔
110	}
111
112	if node.Spec.Unschedulable && ns.excludeUnschedulable {	41✔
113	log.Debugf("Skipping node %s because it is unschedulable", node.Name)	1✔
114	continue	1✔
115	}
116
117	log.Debugf("creating endpoint for node %s", node.Name)	39✔
118		39✔
119	ttl := annotations.TTLFromAnnotations(node.Annotations, fmt.Sprintf("node/%s", node.Name))	39✔
120		39✔
121	addrs := annotations.TargetsFromTargetAnnotation(node.Annotations)	39✔
122		39✔
123	if len(addrs) == 0 {	76✔
124	addrs, err = ns.nodeAddresses(node)	37✔
125	if err != nil {	38✔
126	return nil, fmt.Errorf("failed to get node address from %s: %w", node.Name, err)	1✔
127	}	1✔
128	}
129
130	dnsNames := make(map[string]bool)	38✔
131		38✔
132	if ns.fqdnTemplate != nil {	52✔
133	hostnames, err := fqdn.ExecTemplate(ns.fqdnTemplate, node)	14✔
134	if err != nil {	14✔
135	return nil, err	×
136	}	×
137
138	for _, name := range hostnames {	32✔
139	dnsNames[name] = true	18✔
140	log.Debugf("applied template for %s, converting to %s", node.Name, name)	18✔
141	}	18✔
142	} else {	24✔
143	dnsNames[node.Name] = true	24✔
144	log.Debugf("not applying template for %s", node.Name)	24✔
145	}	24✔
146
147	for dns := range dnsNames {	80✔
148	log.Debugf("adding endpoint with %d targets", len(addrs))	42✔
149		42✔
150	for _, addr := range addrs {	101✔
151	ep := endpoint.NewEndpointWithTTL(dns, suitableType(addr), ttl)	59✔
152	log.Debugf("adding endpoint %s target %s", ep, addr)	59✔
153	key := endpoint.EndpointKey{	59✔
154	DNSName: ep.DNSName,	59✔
155	RecordType: ep.RecordType,	59✔
156	}	59✔
157	if _, ok := endpoints[key]; !ok {	116✔
158	epCopy := *ep	57✔
159	epCopy.RecordType = key.RecordType	57✔
160	endpoints[key] = &epCopy	57✔
161	}	57✔
162	endpoints[key].Targets = append(endpoints[key].Targets, addr)	59✔
163	}
164	}
165	}
166
167	endpointsSlice := []*endpoint.Endpoint{}	38✔
168	for _, ep := range endpoints {	95✔
169	endpointsSlice = append(endpointsSlice, ep)	57✔
170	}	57✔
171
172	return endpointsSlice, nil	38✔
173	}
174
NEW 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) {	37✔
181	addresses := map[v1.NodeAddressType][]string{	37✔
182	v1.NodeExternalIP: {},	37✔
183	v1.NodeInternalIP: {},	37✔
184	}	37✔
185	var internalIpv6Addresses []string	37✔
186		37✔
187	for _, addr := range node.Status.Addresses {	92✔
188	// IPv6 InternalIP addresses have special handling.	55✔
189	// Refer to https://github.com/kubernetes-sigs/external-dns/pull/5192 for more details.	55✔
190	if addr.Type == v1.NodeInternalIP && suitableType(addr.Address) == endpoint.RecordTypeAAAA {	71✔
191	internalIpv6Addresses = append(internalIpv6Addresses, addr.Address)	16✔
192	}	16✔
193	addresses[addr.Type] = append(addresses[addr.Type], addr.Address)	55✔
194	}
195
196	if len(addresses[v1.NodeExternalIP]) > 0 {	62✔
197	if ns.exposeInternalIPv6 {	48✔
198	log.Warn(warningMsg)	23✔
199	return append(addresses[v1.NodeExternalIP], internalIpv6Addresses...), nil	23✔
200	}	23✔
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) {	39✔
213	selector, err := annotations.ParseFilter(ns.annotationFilter)	39✔
214	if err != nil {	39✔
215	return nil, err	×
216	}	×
217
218	// empty filter returns original list
219	if selector.Empty() {	76✔
220	return nodes, nil	37✔
221	}	37✔
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	}

kubernetes-sigs / external-dns / 15497247364

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