14143506289

Committed 29 Mar 2025 07:27AM UTC coverage: 71.618% (+0.3%) from 71.367%

Build # 14143506289

Build Type

Pull #5222

github

Committed by

ivankatliarchuk

Commit Message

Merge remote-tracking branch 'refs/remotes/origin/code-cleanup-main-go' into code-cleanup-main-go

* refs/remotes/origin/code-cleanup-main-go:
  chore(code-cleanup): move logic away from main.go add tests

Pull Request Pull Request #5222: chore(code-cleanup): move logic away from main.go add tests

Run Details

60 of 382 new or added lines in 4 files covered. (15.71%)

13 existing lines in 1 file now uncovered.

14368 of 20062 relevant lines covered (71.62%)

702.88 hits per line

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

90.48

/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"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/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"
)

type nodeSource struct {
        client             kubernetes.Interface
        annotationFilter   string
        fqdnTemplate       *template.Template
        nodeInformer       coreinformers.NodeInformer
        labelSelector      labels.Selector
        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) (Source, error) {
        tmpl, err := 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 := waitForCacheSync(context.Background(), informerFactory); err != nil {
                return nil, err
        }

        return &nodeSource{
                client:             kubeClient,
                annotationFilter:   annotationFilter,
                fqdnTemplate:       tmpl,
                nodeInformer:       nodeInformer,
                labelSelector:      labelSelector,
                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 {
                        log.Debugf("Skipping node %s because it is unschedulable", node.Name)
                        continue
                }

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

                ttl := getTTLFromAnnotations(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 := 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 := getTargetsFromTargetAnnotation(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 {
                        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) {
        labelSelector, err := metav1.ParseToLabelSelector(ns.annotationFilter)
        if err != nil {
                return nil, err
        }
        selector, err := metav1.LabelSelectorAsSelector(labelSelector)
        if err != nil {
                return nil, err
        }

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

        filteredList := []*v1.Node{}

        for _, node := range nodes {
                // convert the node's annotations to an equivalent label selector
                annotations := labels.Set(node.Annotations)

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

kubernetes-sigs / external-dns / 14143506289

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