Skip to content

Move fabric8 discovery to listers #2133

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wind57 wants to merge 31 commits into
base: main
Choose a base branch
from
Draft

Move fabric8 discovery to listers #2133

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
31 commits
Select commit Hold shift + click to select a range
e38ab5f
fix issue
wind57 Jan 30, 2026
abc7c0f
Merge branch 'main' into fix-2072-missing-zone
wind57 Jan 31, 2026
08f0e87
Merge branch 'main' into move-fabric8-discovery-to-listers
wind57 Feb 1, 2026
51864ff
started work
wind57 Feb 1, 2026
ba9b9fc
Merge branch 'remove-double-configuration' into move-fabric8-discover…
wind57 Feb 1, 2026
f02b796
wip
wind57 Feb 1, 2026
7db1fab
Merge branch 'remove-double-configuration' into move-fabric8-discover…
wind57 Feb 1, 2026
bb4ecc6
rename method
wind57 Feb 1, 2026
536db1b
Merge branch 'remove-double-configuration' into move-fabric8-discover…
wind57 Feb 1, 2026
a283a66
Merge branch 'main' into move-fabric8-discovery-to-listers
wind57 Feb 1, 2026
cac69c1
Merge branch 'main' into move-fabric8-discovery-to-listers
wind57 Feb 2, 2026
ee91966
started basic work
wind57 Feb 2, 2026
e4a735c
Merge branch 'move-to-call-generator' into move-fabric8-discovery-to-…
wind57 Feb 3, 2026
b2089f5
wip
wind57 Feb 3, 2026
8b928c0
wip
wind57 Feb 3, 2026
afe758f
wip
wind57 Feb 3, 2026
41d3ae9
wip
wind57 Feb 4, 2026
9389d52
refactored tests
wind57 Feb 4, 2026
3d7cda5
revert test
wind57 Feb 4, 2026
1a84473
wip
wind57 Feb 5, 2026
a1daeee
fix
wind57 Feb 5, 2026
dfe9930
checkstyle
wind57 Feb 5, 2026
6971ed1
Merge branch 'main' into move-fabric8-discovery-to-listers
wind57 Feb 5, 2026
8d618e1
add auto-configuration
wind57 Feb 5, 2026
e839167
checkstyle
wind57 Feb 5, 2026
4b0a9a6
trigger
wind57 Feb 5, 2026
b68952c
trigger
wind57 Feb 5, 2026
b1045d0
fix
wind57 Feb 5, 2026
488e709
Merge branch 'main' into fix-failing-tests
wind57 Feb 6, 2026
552d8a1
merge main
wind57 Feb 6, 2026
18b91cb
Merge branch 'fix-failing-tests' into move-fabric8-discovery-to-listers
wind57 Feb 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 22 additions & 25 deletions ...ava/org/springframework/cloud/kubernetes/commons/leader/election/PodReadyRunnerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,9 +56,8 @@ void readinessOKFromTheFirstCycle(CapturedOutput output) throws Exception {
CompletableFuture<Void> readinessFuture = podReadyRunner.podReady(readinessSupplier);
readinessFuture.get();

assertThat(output.getOut()).contains("Pod : identity in namespace : namespace is ready");
assertThat(output.getOut()).contains("canceling scheduled future because readiness succeeded");

awaitUntil(3, 100, () -> output.getOut().contains("Pod : identity in namespace : namespace is ready"));
awaitUntil(3, 100, () -> output.getOut().contains("canceling scheduled future because readiness succeeded"));
awaitUntil(3, 200, () -> output.getOut().contains("Shutting down executor : podReadyExecutor"));
}

Expand All @@ -80,10 +79,10 @@ void readinessOKFromTheSecondCycle(CapturedOutput output) throws Exception {
CompletableFuture<Void> readinessFuture = podReadyRunner.podReady(readinessSupplier);
readinessFuture.get();

assertThat(output.getOut())
.contains("Pod : identity in namespace : namespace is not ready, will retry in one second");
assertThat(output.getOut()).contains("Pod : identity in namespace : namespace is ready");
assertThat(output.getOut()).contains("canceling scheduled future because readiness succeeded");
awaitUntil(3, 100, () -> output.getOut()
.contains("Pod : identity in namespace : namespace is not ready, will retry in one second"));
awaitUntil(3, 100, () -> output.getOut().contains("Pod : identity in namespace : namespace is ready"));
awaitUntil(3, 100, () -> output.getOut().contains("canceling scheduled future because readiness succeeded"));

awaitUntil(3, 200, () -> output.getOut().contains("Shutting down executor : podReadyExecutor"));
}
Expand All @@ -110,12 +109,11 @@ void readinessFailsOnTheSecondCycle(CapturedOutput output) {
}
catch (Exception e) {
caught = true;
assertThat(output.getOut())
.contains("Pod : identity in namespace : namespace is not ready, will retry in one second");
assertThat(output.getOut()).contains("exception waiting for pod : identity");
assertThat(output.getOut()).contains("pod readiness for : identity failed with : fail on the second cycle");
assertThat(output.getOut()).contains("canceling scheduled future because readiness failed");

awaitUntil(3, 100, () -> output.getOut()
.contains("Pod : identity in namespace : namespace is not ready, will retry in one second"));
awaitUntil(3, 100, () -> output.getOut().contains("exception waiting for pod : identity"));
awaitUntil(3, 100, () -> output.getOut().contains("pod readiness for : identity failed with : fail on the second cycle"));
awaitUntil(3, 100, () -> output.getOut().contains("canceling scheduled future because readiness failed"));
awaitUntil(3, 200, () -> output.getOut().contains("Shutting down executor : podReadyExecutor"));
}
assertThat(caught).isTrue();
Expand Down Expand Up @@ -155,14 +153,13 @@ void readinessFailsOnTheSecondCycleAttachNewPipeline(CapturedOutput output) {
}
catch (Exception e) {
caught = true;
assertThat(output.getOut())
.contains("Pod : identity in namespace : namespace is not ready, will retry in one second");
assertThat(output.getOut()).contains("exception waiting for pod : identity");
assertThat(output.getOut()).contains("pod readiness for : identity failed with : fail on the second cycle");
assertThat(output.getOut()).contains("readiness failed and we caught that");

awaitUntil(3, 100, () -> output.getOut()
.contains("Pod : identity in namespace : namespace is not ready, will retry in one second"));
awaitUntil(3, 100, () -> output.getOut().contains("exception waiting for pod : identity"));
awaitUntil(3, 100, () -> output.getOut().contains("pod readiness for : identity failed with : fail on the second cycle"));
awaitUntil(3, 100, () -> output.getOut().contains("readiness failed and we caught that"));
awaitUntil(3, 200, () -> output.getOut().contains("Shutting down executor : podReadyExecutor"));
assertThat(output.getOut()).contains("canceling scheduled future because readiness failed");
awaitUntil(3, 100, () -> output.getOut().contains("canceling scheduled future because readiness failed"));
}
assertThat(caught).isTrue();
}
Expand Down Expand Up @@ -210,14 +207,14 @@ void readinessCanceledOnTheSecondCycleAttachNewPipeline(CapturedOutput output) t
assertThat(output.getOut())
.contains("Pod : identity in namespace : namespace is not ready, will retry in one second");
// this is a cancel of the future, not an exception per se
assertThat(output.getOut()).doesNotContain("leader election for : identity was not successful");
assertThat(output.getOut()).contains("readiness failed and we caught that");
awaitUntil(3, 100, () -> !output.getOut().contains("leader election for : identity was not successful"));
awaitUntil(3, 100, () -> output.getOut().contains("readiness failed and we caught that"));

awaitUntil(3, 200, () -> output.getOut().contains("Shutting down executor : podReadyExecutor"));

assertThat(output.getOut()).contains("canceling scheduled future because completable future was cancelled");
assertThat(output.getOut()).doesNotContain("canceling scheduled future because readiness failed");
assertThat(output.getOut()).contains("scheduledFuture is canceled: true");
awaitUntil(3, 100, () -> output.getOut().contains("canceling scheduled future because completable future was cancelled"));
awaitUntil(3, 100, () -> !output.getOut().contains("canceling scheduled future because readiness failed"));
awaitUntil(3, 100, () -> output.getOut().contains("scheduledFuture is canceled: true"));

}
assertThat(caught).isTrue();
Expand Down
8 changes: 8 additions & 0 deletions .../src/main/java/org/springframework/cloud/kubernetes/fabric8/Fabric8AutoConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,10 @@
import org.springframework.boot.cloud.CloudPlatform;
import org.springframework.cloud.kubernetes.commons.KubernetesClientProperties;
import org.springframework.cloud.kubernetes.commons.KubernetesCommonsAutoConfiguration;
import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

/**
* Auto configuration for Kubernetes.
Expand Down Expand Up @@ -105,6 +107,12 @@ Fabric8PodUtils kubernetesPodUtils(KubernetesClient client) {
return new Fabric8PodUtils(client);
}

@Bean
@ConditionalOnMissingBean
Copy link
Contributor Author

@wind57 wind57 Feb 5, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we used to create this one in place where needed, but then I noticed that in the k8s-client native, we create in the auto-configuration. So I decided to do it in the fabric8 case too.

Also, it simplifies testing a lot, having it as a Bean

KubernetesNamespaceProvider kubernetesNamespaceProvider(Environment environment) {
return new KubernetesNamespaceProvider(environment);
}

private static <D> D or(D left, D right) {
return left != null ? left : right;
}
Expand Down
175 changes: 115 additions & 60 deletions ...gframework/cloud/kubernetes/fabric8/discovery/Fabric8AbstractBlockingDiscoveryClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,17 +21,22 @@
import java.util.Map;
import java.util.Objects;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import io.fabric8.kubernetes.api.model.EndpointAddress;
import io.fabric8.kubernetes.api.model.EndpointSubset;
import io.fabric8.kubernetes.api.model.Endpoints;
import io.fabric8.kubernetes.api.model.Service;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
import io.fabric8.kubernetes.client.informers.cache.Lister;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import org.apache.commons.logging.LogFactory;

import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
import org.springframework.cloud.kubernetes.commons.discovery.ServiceMetadata;
import org.springframework.cloud.kubernetes.commons.discovery.ServicePortNameAndNumber;
Expand All @@ -46,10 +51,9 @@
import static org.springframework.cloud.kubernetes.fabric8.Fabric8Utils.serviceMetadata;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8DiscoveryClientUtils.addresses;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8DiscoveryClientUtils.endpointSubsetsPortData;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8DiscoveryClientUtils.endpoints;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8DiscoveryClientUtils.services;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8InstanceIdHostPodNameSupplier.externalName;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8InstanceIdHostPodNameSupplier.nonExternalName;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8DiscoveryClientUtils.postConstruct;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8InstanceIdHostPodNameSupplier.fabric8InstanceIdHostPodNameSupplier;
import static org.springframework.cloud.kubernetes.fabric8.discovery.Fabric8PodLabelsAndAnnotationsSupplier.fabric8PodLabelsAndAnnotationsSupplier;

/**
* @author wind57
Expand All @@ -59,70 +63,97 @@ abstract class Fabric8AbstractBlockingDiscoveryClient implements DiscoveryClient
private static final LogAccessor LOG = new LogAccessor(
LogFactory.getLog(Fabric8AbstractBlockingDiscoveryClient.class));

private final KubernetesDiscoveryProperties properties;
private final List<Lister<Service>> serviceListers;
Copy link
Contributor Author

@wind57 wind57 Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Listers and Informers here, just like the case for the k8s-native client. Unlike the native client, we don't have to deal with SharedInfomerFactory, so a bit simpler


private final ServicePortSecureResolver servicePortSecureResolver;
private final List<Lister<Endpoints>> endpointsListers;

private final KubernetesClient client;
private final List<SharedIndexInformer<Service>> serviceInformers;

private final KubernetesNamespaceProvider namespaceProvider;
private final List<SharedIndexInformer<Endpoints>> endpointsInformers;

private final Supplier<Boolean> informersReadyFunc;

private final KubernetesDiscoveryProperties properties;

private final Predicate<Service> predicate;

Fabric8AbstractBlockingDiscoveryClient(KubernetesClient client,
KubernetesDiscoveryProperties kubernetesDiscoveryProperties,
ServicePortSecureResolver servicePortSecureResolver, KubernetesNamespaceProvider namespaceProvider,
private final ServicePortSecureResolver servicePortSecureResolver;

private final KubernetesClient kubernetesClient;

Fabric8AbstractBlockingDiscoveryClient(KubernetesClient kubernetesClient, List<Lister<Service>> serviceListers,
List<Lister<Endpoints>> endpointsListers, List<SharedIndexInformer<Service>> serviceInformers,
List<SharedIndexInformer<Endpoints>> endpointsInformers, KubernetesDiscoveryProperties properties,
Predicate<Service> predicate) {

this.client = client;
this.properties = kubernetesDiscoveryProperties;
this.servicePortSecureResolver = servicePortSecureResolver;
this.namespaceProvider = namespaceProvider;
this.serviceListers = serviceListers;
this.endpointsListers = endpointsListers;
this.serviceInformers = serviceInformers;
this.endpointsInformers = endpointsInformers;
this.properties = properties;
this.predicate = predicate;
this.kubernetesClient = kubernetesClient;

servicePortSecureResolver = new ServicePortSecureResolver(properties);

this.informersReadyFunc = () -> {
boolean serviceInformersReady = serviceInformers.isEmpty() || serviceInformers.stream()
.map(SharedIndexInformer::hasSynced)
.reduce(Boolean::logicalAnd)
.orElse(false);
boolean endpointsInformersReady = endpointsInformers.isEmpty() || endpointsInformers.stream()
.map(SharedIndexInformer::hasSynced)
.reduce(Boolean::logicalAnd)
.orElse(false);
return serviceInformersReady && endpointsInformersReady;
};
}

public abstract String description();

@Override
public List<ServiceInstance> getInstances(String serviceId) {
Objects.requireNonNull(serviceId);
Objects.requireNonNull(serviceId, "serviceId must be provided");

List<Endpoints> allEndpoints = endpoints(properties, client, namespaceProvider, "fabric8-discovery", serviceId,
predicate);
List<Service> allServices = serviceListers.stream()
Copy link
Contributor Author

@wind57 wind57 Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is a 1-1 mapping with k8s-native, meaning we do things the same here, much easier to reason like this

.flatMap(x -> x.list().stream())
.filter(service -> service.getMetadata() != null)
.filter(service -> serviceId.equals(service.getMetadata().getName()))
.toList();

List<ServiceInstance> instances = new ArrayList<>();
for (Endpoints endpoints : allEndpoints) {
// endpoints are only those that matched the serviceId
instances.addAll(serviceInstances(endpoints, serviceId));
}
List<ServiceInstance> serviceInstances = allServices.stream()
.filter(predicate)
.flatMap(service -> serviceInstances(service).stream())
.collect(Collectors.toCollection(ArrayList::new));

if (properties.includeExternalNameServices()) {
LOG.debug(() -> "Searching for 'ExternalName' type of services with serviceId : " + serviceId);
List<Service> services = services(properties, client, namespaceProvider,
s -> s.getSpec().getType().equals(EXTERNAL_NAME), Map.of("metadata.name", serviceId),
"fabric8-discovery");
for (Service service : services) {
List<Service> externalNameServices = allServices.stream()
.filter(s -> s.getSpec() != null)
.filter(s -> EXTERNAL_NAME.equals(s.getSpec().getType()))
.toList();
for (Service service : externalNameServices) {
ServiceMetadata serviceMetadata = serviceMetadata(service);
Map<String, String> serviceInstanceMetadata = serviceInstanceMetadata(Map.of(), serviceMetadata,
properties);

Fabric8InstanceIdHostPodNameSupplier supplierOne = externalName(service);

ServiceInstance externalNameServiceInstance = externalNameServiceInstance(serviceMetadata, supplierOne,
serviceInstanceMetadata);

instances.add(externalNameServiceInstance);
Fabric8InstanceIdHostPodNameSupplier fabric8InstanceIdHostPodNameSupplier = fabric8InstanceIdHostPodNameSupplier(
service);
ServiceInstance externalNameServiceInstance = externalNameServiceInstance(serviceMetadata,
fabric8InstanceIdHostPodNameSupplier, serviceInstanceMetadata);
serviceInstances.add(externalNameServiceInstance);
}
}

return instances;
return serviceInstances;
}

@Override
public List<String> getServices() {
List<String> services = services(properties, client, namespaceProvider, predicate, null, "fabric8 discovery")
.stream()
.map(service -> service.getMetadata().getName())
List<String> services = serviceListers.stream()
.flatMap(serviceLister -> serviceLister.list().stream())
.filter(predicate)
.map(s -> s.getMetadata().getName())
.distinct()
.toList();
LOG.debug(() -> "will return services : " + services);
Expand All @@ -134,38 +165,62 @@ public int getOrder() {
return properties.order();
}

private List<ServiceInstance> serviceInstances(Endpoints endpoints, String serviceId) {
@PostConstruct
void afterPropertiesSet() {
postConstruct(properties, informersReadyFunc, serviceListers);
}

List<EndpointSubset> subsets = endpoints.getSubsets();
if (subsets.isEmpty()) {
LOG.debug(() -> "serviceId : " + serviceId + " does not have any subsets");
return List.of();
}
@PreDestroy
void preDestroy() {
serviceInformers.forEach(SharedIndexInformer::close);
endpointsInformers.forEach(SharedIndexInformer::close);
}

private List<ServiceInstance> serviceInstances(Service service) {

String serviceId = service.getMetadata().getName();
String serviceNamespace = service.getMetadata().getNamespace();

String namespace = endpoints.getMetadata().getNamespace();
List<ServiceInstance> instances = new ArrayList<>();

Service service = client.services().inNamespace(namespace).withName(serviceId).get();
ServiceMetadata serviceMetadata = serviceMetadata(service);
Map<String, Integer> portsData = endpointSubsetsPortData(subsets);
List<Endpoints> allEndpoints = endpointsListers.stream()
.map(endpointsLister -> endpointsLister.namespace(serviceNamespace).get(serviceId))
.filter(Objects::nonNull)
.toList();

ServiceMetadata k8sServiceMetadata = serviceMetadata(service);

for (Endpoints endpoints : allEndpoints) {
List<EndpointSubset> subsets = endpoints.getSubsets();
if (subsets == null || subsets.isEmpty()) {
LOG.debug(() -> "serviceId : " + serviceId + " does not have any subsets");
}
else {
Map<String, Integer> portsData = endpointSubsetsPortData(subsets);
Map<String, String> serviceInstanceMetadata = serviceInstanceMetadata(portsData, k8sServiceMetadata,
properties);

Map<String, String> serviceInstanceMetadata = serviceInstanceMetadata(portsData, serviceMetadata, properties);
for (EndpointSubset endpointSubset : subsets) {

for (EndpointSubset endpointSubset : subsets) {
Map<String, Integer> endpointsPortData = endpointSubsetsPortData(List.of(endpointSubset));
ServicePortNameAndNumber portData = endpointsPort(endpointsPortData, k8sServiceMetadata,
properties);

Map<String, Integer> endpointsPortData = endpointSubsetsPortData(List.of(endpointSubset));
ServicePortNameAndNumber portData = endpointsPort(endpointsPortData, serviceMetadata, properties);
List<EndpointAddress> addresses = addresses(endpointSubset, properties);
for (EndpointAddress endpointAddress : addresses) {

List<EndpointAddress> addresses = addresses(endpointSubset, properties);
for (EndpointAddress endpointAddress : addresses) {
Fabric8InstanceIdHostPodNameSupplier instanceIdHostPodNameSupplier = fabric8InstanceIdHostPodNameSupplier(
endpointAddress, service);
Fabric8PodLabelsAndAnnotationsSupplier podLabelsAndAnnotationsSupplier = fabric8PodLabelsAndAnnotationsSupplier(
kubernetesClient, service.getMetadata().getNamespace());

Fabric8InstanceIdHostPodNameSupplier supplierOne = nonExternalName(endpointAddress, service);
Fabric8PodLabelsAndAnnotationsSupplier supplierTwo = Fabric8PodLabelsAndAnnotationsSupplier
.nonExternalName(client, namespace);
ServiceInstance serviceInstance = serviceInstance(servicePortSecureResolver, k8sServiceMetadata,
instanceIdHostPodNameSupplier, podLabelsAndAnnotationsSupplier, portData,
serviceInstanceMetadata, properties);
instances.add(serviceInstance);
}
}

ServiceInstance serviceInstance = serviceInstance(servicePortSecureResolver, serviceMetadata,
supplierOne, supplierTwo, portData, serviceInstanceMetadata, properties);
instances.add(serviceInstance);
}
}

Expand Down
Loading
Loading