Skip to content

Commit 6cf328f

Browse files
AndrewCharlesHaymloiseleur
AndrewCharlesHay
and
mloiseleur
authored
refactor(service): reduce cyclomatic complexity of extractHeadlessEndpoints (#5822)
* feat: reduce cyclomatic complexity of service_test * style: indention added * style: tab * refactor: address PR feedback, improve tests, and reduce complexity * fix(service): address PR feedback and fix linting * Revert "fix(service): address PR feedback and fix linting" This reverts commit 4cba488. * refactor: address all PR feedback - improve method naming, add test coverage, fix parameter ordering * refactor: address latest PR feedback - convert to method, use testutils, add test coverage * test: add coverage for pod hostname scenario Addresses 4th to last PR comment about missing test coverage for the case where pod.Spec.Hostname is set, which creates additional headless domains (pod-specific hostname + base hostname) * style: remove extra line Co-authored-by: Michel Loiseleur <[email protected]> * style: remove extra line Co-authored-by: Michel Loiseleur <[email protected]> --------- Co-authored-by: Michel Loiseleur <[email protected]>
1 parent fe753cb commit 6cf328f

File tree

3 files changed

+566
-72
lines changed

3 files changed

+566
-72
lines changed

‎.golangci.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ linters:
4141
- name: confusing-naming
4242
disabled: true
4343
cyclop: # Lower cyclomatic complexity threshold after the max complexity is lowered
44-
max-complexity: 44
44+
max-complexity: 43
4545
testifylint:
4646
# Enable all checkers (https://github.com/Antonboom/testifylint#checkers).
4747
# Default: false

‎source/service.go‎

Lines changed: 102 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -331,22 +331,11 @@ func (sc *serviceSource) extractHeadlessEndpoints(svc *v1.Service, hostname stri
331331
serviceKey := cache.ObjectName{Namespace: svc.Namespace, Name: svc.Name}.String()
332332
rawEndpointSlices, err := sc.endpointSlicesInformer.Informer().GetIndexer().ByIndex(serviceNameIndexKey, serviceKey)
333333
if err != nil {
334-
// Should never happen as long as the index exists
335334
log.Errorf("Get EndpointSlices of service[%s] error:%v", svc.GetName(), err)
336335
return nil
337336
}
338337

339-
endpointSlices := make([]*discoveryv1.EndpointSlice, 0, len(rawEndpointSlices))
340-
for _, obj := range rawEndpointSlices {
341-
endpointSlice, ok := obj.(*discoveryv1.EndpointSlice)
342-
if !ok {
343-
// Should never happen as the indexer can only contain EndpointSlice objects
344-
log.Errorf("Expected %T but got %T instead, skipping", endpointSlice, obj)
345-
continue
346-
}
347-
endpointSlices = append(endpointSlices, endpointSlice)
348-
}
349-
338+
endpointSlices := convertToEndpointSlices(rawEndpointSlices)
350339
pods, err := sc.podInformer.Lister().Pods(svc.Namespace).List(selector)
351340
if err != nil {
352341
log.Errorf("List Pods of service[%s] error:%v", svc.GetName(), err)
@@ -357,74 +346,63 @@ func (sc *serviceSource) extractHeadlessEndpoints(svc *v1.Service, hostname stri
357346
publishPodIPs := endpointsType != EndpointsTypeNodeExternalIP && endpointsType != EndpointsTypeHostIP && !sc.publishHostIP
358347
publishNotReadyAddresses := svc.Spec.PublishNotReadyAddresses || sc.alwaysPublishNotReadyAddresses
359348

349+
targetsByHeadlessDomainAndType := sc.processHeadlessEndpointsFromSlices(
350+
svc, pods, endpointSlices, hostname, endpointsType, publishPodIPs, publishNotReadyAddresses)
351+
endpoints = buildHeadlessEndpoints(svc, targetsByHeadlessDomainAndType, ttl)
352+
return endpoints
353+
}
354+
355+
// Helper to convert raw objects to EndpointSlice
356+
func convertToEndpointSlices(rawEndpointSlices []interface{}) []*discoveryv1.EndpointSlice {
357+
endpointSlices := make([]*discoveryv1.EndpointSlice, 0, len(rawEndpointSlices))
358+
for _, obj := range rawEndpointSlices {
359+
endpointSlice, ok := obj.(*discoveryv1.EndpointSlice)
360+
if !ok {
361+
log.Errorf("Expected EndpointSlice but got %T instead, skipping", obj)
362+
continue
363+
}
364+
endpointSlices = append(endpointSlices, endpointSlice)
365+
}
366+
return endpointSlices
367+
}
368+
369+
// processHeadlessEndpointsFromSlices processes EndpointSlices specifically for headless services
370+
// and returns deduped targets by domain/type.
371+
// TODO: Consider refactoring with generics when available: https://github.com/kubernetes/kubernetes/issues/133544
372+
func (sc *serviceSource) processHeadlessEndpointsFromSlices(
373+
svc *v1.Service,
374+
pods []*v1.Pod,
375+
endpointSlices []*discoveryv1.EndpointSlice,
376+
hostname string,
377+
endpointsType string,
378+
publishPodIPs bool,
379+
publishNotReadyAddresses bool,
380+
) map[endpoint.EndpointKey]endpoint.Targets {
360381
targetsByHeadlessDomainAndType := make(map[endpoint.EndpointKey]endpoint.Targets)
361382
for _, endpointSlice := range endpointSlices {
362383
for _, ep := range endpointSlice.Endpoints {
363384
if !conditionToBool(ep.Conditions.Ready) && !publishNotReadyAddresses {
364385
continue
365386
}
366-
367-
if publishPodIPs &&
368-
endpointSlice.AddressType != discoveryv1.AddressTypeIPv4 &&
369-
endpointSlice.AddressType != discoveryv1.AddressTypeIPv6 {
387+
if publishPodIPs && endpointSlice.AddressType != discoveryv1.AddressTypeIPv4 && endpointSlice.AddressType != discoveryv1.AddressTypeIPv6 {
370388
log.Debugf("Skipping EndpointSlice %s/%s because its address type is unsupported: %s", endpointSlice.Namespace, endpointSlice.Name, endpointSlice.AddressType)
371389
continue
372390
}
373-
374-
// find pod for this address
375-
if ep.TargetRef == nil || ep.TargetRef.APIVersion != "" || ep.TargetRef.Kind != "Pod" {
376-
log.Debugf("Skipping address because its target is not a pod: %v", ep)
377-
continue
378-
}
379-
var pod *v1.Pod
380-
for _, v := range pods {
381-
if v.Name == ep.TargetRef.Name {
382-
pod = v
383-
break
384-
}
385-
}
391+
pod := findPodForEndpoint(ep, pods)
386392
if pod == nil {
387-
log.Errorf("Pod %s not found for address %v", ep.TargetRef.Name, ep)
393+
if ep.TargetRef != nil {
394+
log.Errorf("Pod %s not found for address %v", ep.TargetRef.Name, ep)
395+
} else {
396+
log.Errorf("Pod not found for endpoint with nil TargetRef: %v", ep)
397+
}
388398
continue
389399
}
390-
391400
headlessDomains := []string{hostname}
392401
if pod.Spec.Hostname != "" {
393402
headlessDomains = append(headlessDomains, fmt.Sprintf("%s.%s", pod.Spec.Hostname, hostname))
394403
}
395-
396404
for _, headlessDomain := range headlessDomains {
397-
targets := annotations.TargetsFromTargetAnnotation(pod.Annotations)
398-
if len(targets) == 0 {
399-
if endpointsType == EndpointsTypeNodeExternalIP {
400-
if sc.nodeInformer == nil {
401-
log.Warnf("Skipping EndpointSlice %s/%s as --service-type-filter disable node informer", endpointSlice.Namespace, endpointSlice.Name)
402-
continue
403-
}
404-
node, err := sc.nodeInformer.Lister().Get(pod.Spec.NodeName)
405-
if err != nil {
406-
log.Errorf("Get node[%s] of pod[%s] error: %v; not adding any NodeExternalIP endpoints", pod.Spec.NodeName, pod.GetName(), err)
407-
return endpoints
408-
}
409-
for _, address := range node.Status.Addresses {
410-
if address.Type == v1.NodeExternalIP || (sc.exposeInternalIPv6 && address.Type == v1.NodeInternalIP && suitableType(address.Address) == endpoint.RecordTypeAAAA) {
411-
targets = append(targets, address.Address)
412-
log.Debugf("Generating matching endpoint %s with NodeExternalIP %s", headlessDomain, address.Address)
413-
}
414-
}
415-
} else if endpointsType == EndpointsTypeHostIP || sc.publishHostIP {
416-
targets = endpoint.Targets{pod.Status.HostIP}
417-
log.Debugf("Generating matching endpoint %s with HostIP %s", headlessDomain, pod.Status.HostIP)
418-
} else {
419-
if len(ep.Addresses) == 0 {
420-
log.Warnf("EndpointSlice %s/%s has no addresses for endpoint %v", endpointSlice.Namespace, endpointSlice.Name, ep)
421-
continue
422-
}
423-
address := ep.Addresses[0] // Only use the first address, as additional addresses have no semantic defined
424-
targets = endpoint.Targets{address}
425-
log.Debugf("Generating matching endpoint %s with EndpointSliceAddress IP %s", headlessDomain, address)
426-
}
427-
}
405+
targets := sc.getTargetsForDomain(pod, ep, endpointSlice, endpointsType, headlessDomain)
428406
for _, target := range targets {
429407
key := endpoint.EndpointKey{
430408
DNSName: headlessDomain,
@@ -435,8 +413,68 @@ func (sc *serviceSource) extractHeadlessEndpoints(svc *v1.Service, hostname stri
435413
}
436414
}
437415
}
416+
// Return a copy of the map to prevent external modifications
417+
result := make(map[endpoint.EndpointKey]endpoint.Targets, len(targetsByHeadlessDomainAndType))
418+
for k, v := range targetsByHeadlessDomainAndType {
419+
result[k] = append(endpoint.Targets(nil), v...)
420+
}
421+
return result
422+
}
423+
424+
// Helper to find pod for endpoint
425+
func findPodForEndpoint(ep discoveryv1.Endpoint, pods []*v1.Pod) *v1.Pod {
426+
if ep.TargetRef == nil || ep.TargetRef.APIVersion != "" || ep.TargetRef.Kind != "Pod" {
427+
log.Debugf("Skipping address because its target is not a pod: %v", ep)
428+
return nil
429+
}
430+
for _, v := range pods {
431+
if v.Name == ep.TargetRef.Name {
432+
return v
433+
}
434+
}
435+
return nil
436+
}
438437

439-
headlessKeys := []endpoint.EndpointKey{}
438+
// Helper to get targets for domain
439+
func (sc *serviceSource) getTargetsForDomain(pod *v1.Pod, ep discoveryv1.Endpoint, endpointSlice *discoveryv1.EndpointSlice, endpointsType string, headlessDomain string) endpoint.Targets {
440+
targets := annotations.TargetsFromTargetAnnotation(pod.Annotations)
441+
if len(targets) == 0 {
442+
if endpointsType == EndpointsTypeNodeExternalIP {
443+
if sc.nodeInformer == nil {
444+
log.Warnf("Skipping EndpointSlice %s/%s as --service-type-filter disable node informer", endpointSlice.Namespace, endpointSlice.Name)
445+
return nil
446+
}
447+
node, err := sc.nodeInformer.Lister().Get(pod.Spec.NodeName)
448+
if err != nil {
449+
log.Errorf("Get node[%s] of pod[%s] error: %v; not adding any NodeExternalIP endpoints", pod.Spec.NodeName, pod.GetName(), err)
450+
return nil
451+
}
452+
for _, address := range node.Status.Addresses {
453+
if address.Type == v1.NodeExternalIP || (sc.exposeInternalIPv6 && address.Type == v1.NodeInternalIP && suitableType(address.Address) == endpoint.RecordTypeAAAA) {
454+
targets = append(targets, address.Address)
455+
log.Debugf("Generating matching endpoint %s with NodeExternalIP %s", headlessDomain, address.Address)
456+
}
457+
}
458+
} else if endpointsType == EndpointsTypeHostIP || sc.publishHostIP {
459+
targets = endpoint.Targets{pod.Status.HostIP}
460+
log.Debugf("Generating matching endpoint %s with HostIP %s", headlessDomain, pod.Status.HostIP)
461+
} else {
462+
if len(ep.Addresses) == 0 {
463+
log.Warnf("EndpointSlice %s/%s has no addresses for endpoint %v", endpointSlice.Namespace, endpointSlice.Name, ep)
464+
return nil
465+
}
466+
address := ep.Addresses[0]
467+
targets = endpoint.Targets{address}
468+
log.Debugf("Generating matching endpoint %s with EndpointSliceAddress IP %s", headlessDomain, address)
469+
}
470+
}
471+
return targets
472+
}
473+
474+
// Helper to build endpoints from deduped targets
475+
func buildHeadlessEndpoints(svc *v1.Service, targetsByHeadlessDomainAndType map[endpoint.EndpointKey]endpoint.Targets, ttl endpoint.TTL) []*endpoint.Endpoint {
476+
var endpoints []*endpoint.Endpoint
477+
headlessKeys := make([]endpoint.EndpointKey, 0, len(targetsByHeadlessDomainAndType))
440478
for headlessKey := range targetsByHeadlessDomainAndType {
441479
headlessKeys = append(headlessKeys, headlessKey)
442480
}
@@ -449,31 +487,26 @@ func (sc *serviceSource) extractHeadlessEndpoints(svc *v1.Service, hostname stri
449487
for _, headlessKey := range headlessKeys {
450488
allTargets := targetsByHeadlessDomainAndType[headlessKey]
451489
targets := []string{}
452-
453490
deduppedTargets := map[string]struct{}{}
454491
for _, target := range allTargets {
455492
if _, ok := deduppedTargets[target]; ok {
456493
log.Debugf("Removing duplicate target %s", target)
457494
continue
458495
}
459-
460496
deduppedTargets[target] = struct{}{}
461497
targets = append(targets, target)
462498
}
463-
464499
var ep *endpoint.Endpoint
465500
if ttl.IsConfigured() {
466501
ep = endpoint.NewEndpointWithTTL(headlessKey.DNSName, headlessKey.RecordType, ttl, targets...)
467502
} else {
468503
ep = endpoint.NewEndpoint(headlessKey.DNSName, headlessKey.RecordType, targets...)
469504
}
470-
471505
if ep != nil {
472506
ep.WithLabel(endpoint.ResourceLabelKey, fmt.Sprintf("service/%s/%s", svc.Namespace, svc.Name))
473507
endpoints = append(endpoints, ep)
474508
}
475509
}
476-
477510
return endpoints
478511
}
479512

0 commit comments

Comments
 (0)