15063209625

Committed 16 May 2025 07:34AM UTC coverage: 73.121% (+0.8%) from 72.283%

Build # 15063209625

Build Type

Pull #5353

github

Committed by

ivankatliarchuk

Commit Message

chore(docs): update aws permissions

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

Pull Request Pull Request #5353: chore(docs): update aws role requirements with conditions

Run Details

14859 of 20321 relevant lines covered (73.12%)

694.09 hits per line

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

91.49

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

kubernetes-sigs / external-dns / 15063209625

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