14314497835

Committed 07 Apr 2025 04:30PM UTC coverage: 72.07% (-0.001%) from 72.071%

Build # 14314497835

Build Type

Pull #5250

github

Committed by

web-flow

Commit Message

chore(helm): add validation for prefix and suffix and capture regression

Co-authored-by: Steve Hipwell <steve.hipwell@gmail.com>

Pull Request Pull Request #5250: chore(helm): add validation for prefix and suffix and capture regression

Run Details

14667 of 20351 relevant lines covered (72.07%)

693.6 hits per line

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

90.54

/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"
)

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

kubernetes-sigs / external-dns / 14314497835

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