16320352320

Committed 16 Jul 2025 01:08PM UTC coverage: 77.209% (+0.02%) from 77.187%

Build # 16320352320

Build Type

Pull #5654

github

Committed by

AndrewCharlesHay

Commit Message

style: remove trailing whitespace

Pull Request Pull Request #5654: chore(cloudflare): use lib v4 for zone services

Run Details

31 of 39 new or added lines in 1 file covered. (79.49%)

19 existing lines in 3 files now uncovered.

14770 of 19130 relevant lines covered (77.21%)

757.41 hits per line

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

91.89

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

kubernetes-sigs / external-dns / 16320352320

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