Skip to content

Commit 0bcd042

Browse files
PalakisPalakis
authored
Service selection with equality-based filtering (#2)
1 parent ae95b99 commit 0bcd042

File tree

4 files changed

+334
-40
lines changed

4 files changed

+334
-40
lines changed

‎src/main/java/com/dtforce/spring/kubernetes/KubernetesDiscoveryClient.java‎

Lines changed: 86 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,13 @@
88
import org.springframework.cloud.client.DefaultServiceInstance;
99
import org.springframework.cloud.client.ServiceInstance;
1010
import org.springframework.cloud.client.discovery.DiscoveryClient;
11-
import org.springframework.util.Assert;
1211

13-
import java.util.ArrayList;
14-
import java.util.Collections;
15-
import java.util.List;
16-
import java.util.Optional;
12+
import java.util.*;
1713

18-
public class KubernetesDiscoveryClient implements DiscoveryClient
14+
public class KubernetesDiscoveryClient implements DiscoveryClient, SelectorEnabledDiscoveryClient
1915
{
16+
private static final String defaultPortName = "http";
17+
2018
private static Logger log = LoggerFactory.getLogger(KubernetesDiscoveryClient.class.getName());
2119

2220
private KubernetesClient kubeClient;
@@ -37,45 +35,103 @@ public List<ServiceInstance> getInstances(String serviceId)
3735
{
3836
Service service;
3937
try {
40-
// API calls are automatically namespaced to the client's assigned namespace.
4138
service = kubeClient.services().withName(serviceId).get();
4239
} catch(KubernetesClientException e) {
43-
log.warn("getInstances: failed to retrieve service '{}': API call failed: {}", serviceId, e.getMessage());
44-
return Collections.emptyList();
40+
log.error("getInstances: failed to retrieve service '{}': API call failed. " +
41+
"Check your K8s client configuration and account permissions.", serviceId);
42+
throw e;
4543
}
4644

45+
// A get() return value can be null, unlike one from a list() call
4746
if (service == null) {
4847
log.warn("getInstances: specified service '{}' doesn't exist", serviceId);
4948
return Collections.emptyList();
5049
}
5150

52-
if (log.isDebugEnabled()) {
53-
log.debug("getInstances: service = {}", service.toString());
51+
return getInstancesFromService(service);
52+
}
53+
54+
@Override
55+
public List<ServiceInstance> selectInstances(Map<String, String> match)
56+
{
57+
return selectInstances(match, Collections.emptyMap());
58+
}
59+
60+
@Override
61+
public List<ServiceInstance> selectInstances(Map<String, String> match, Map<String, String> doNotMatch)
62+
{
63+
ServiceList serviceList;
64+
try {
65+
serviceList = kubeClient.services().withLabels(match).withoutLabels(doNotMatch).list();
66+
} catch(KubernetesClientException e) {
67+
log.error("selectInstances: failed to retrieve matching services: API call failed. " +
68+
"Check your K8s client configuration and account permissions.");
69+
throw e;
5470
}
5571

56-
if (service.getSpec().getPorts().isEmpty()) {
57-
log.error("getInstances: service '{}' has no ports", serviceId);
72+
// serviceList is never supposed to be null, even if the query has no results
73+
if (serviceList == null) {
74+
log.error("selectInstances: service list is null");
5875
return Collections.emptyList();
5976
}
6077

61-
List<ServicePort> servicePorts = service.getSpec().getPorts();
62-
ServicePort svcPort = servicePorts.get(0); // By default, use the first port available
78+
return getInstancesFromServiceList(serviceList.getItems());
79+
}
6380

64-
// But if a port named "http" is available, use it instead
65-
Optional<ServicePort> httpPort = servicePorts.stream()
66-
.filter(s -> s.getName() != null && s.getName().equals("http")).findFirst();
67-
if (httpPort.isPresent()) {
68-
svcPort = httpPort.get();
81+
private List<ServiceInstance> getInstancesFromService(Service service)
82+
{
83+
return getInstancesFromServiceList(Collections.singletonList(service));
84+
}
85+
86+
private List<ServiceInstance> getInstancesFromServiceList(List<Service> services)
87+
{
88+
if (log.isDebugEnabled()) {
89+
log.debug("getInstancesFromServiceList: services = {}", services.toString());
6990
}
7091

7192
List<ServiceInstance> serviceInstances = new ArrayList<>();
72-
serviceInstances.add(new DefaultServiceInstance(
73-
service.getMetadata().getName(),
74-
service.getSpec().getClusterIP(),
75-
svcPort.getPort(),
76-
false,
77-
service.getMetadata().getLabels()
78-
));
93+
for (Service service : services) {
94+
if (service.getSpec() == null) {
95+
log.error("skipping service with no spec");
96+
continue;
97+
}
98+
99+
String serviceName = service.getMetadata().getName();
100+
List<ServicePort> servicePorts = service.getSpec().getPorts();
101+
102+
if (servicePorts.isEmpty()) {
103+
log.error("service '{}' has no ports", serviceName);
104+
continue;
105+
}
106+
107+
ServicePort svcPort;
108+
if (servicePorts.size() > 1) {
109+
Optional<ServicePort> httpPort = servicePorts.stream()
110+
.filter(s -> s.getName() != null && s.getName().equals(defaultPortName))
111+
.findFirst();
112+
113+
if (httpPort.isPresent()) {
114+
svcPort = httpPort.get();
115+
} else {
116+
log.warn("getInstancesFromServiceList: multiple ports detected in '{}' " +
117+
"and named default port '{}' not found. Falling back to first port available.",
118+
serviceName, defaultPortName);
119+
svcPort = servicePorts.get(0);
120+
}
121+
} else {
122+
svcPort = servicePorts.get(0);
123+
}
124+
125+
assert svcPort != null;
126+
127+
serviceInstances.add(new DefaultServiceInstance(
128+
service.getMetadata().getName(),
129+
service.getSpec().getClusterIP(),
130+
svcPort.getPort(),
131+
false,
132+
service.getMetadata().getLabels()
133+
));
134+
}
79135
return serviceInstances;
80136
}
81137

@@ -86,8 +142,9 @@ public List<String> getServices()
86142
try {
87143
serviceList = kubeClient.services().list();
88144
} catch (KubernetesClientException e) {
89-
log.warn("getServices: failed to retrieve the list of services: API call failed: {}", e.getMessage());
90-
return Collections.emptyList();
145+
log.error("getServices: failed to retrieve the list of services: API call failed. " +
146+
"Check your K8s client configuration and account permissions.");
147+
throw e;
91148
}
92149

93150
List<String> serviceNames = new ArrayList<>();

‎src/main/java/com/dtforce/spring/kubernetes/SelectorEnabledDiscoveryClient.java‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
package com.dtforce.spring.kubernetes;
2+
3+
import org.springframework.cloud.client.ServiceInstance;
4+
import org.springframework.cloud.client.discovery.DiscoveryClient;
5+
6+
import java.util.List;
7+
import java.util.Map;
8+
9+
public interface SelectorEnabledDiscoveryClient extends DiscoveryClient
10+
{
11+
12+
/**
13+
* Get instances having the specified metadata pairs (equality-based query)
14+
* @param match
15+
* @return List of service instances matching the query
16+
*/
17+
List<ServiceInstance> selectInstances(Map<String, String> match);
18+
19+
20+
/**
21+
* Get instances having the metadata pairs specified in {@code match} but not
22+
* the ones specified in {@code doNotMatch}
23+
* @param match
24+
* @param doNotMatch
25+
* @return List of service instances matching the query
26+
*/
27+
List<ServiceInstance> selectInstances(Map<String, String> match, Map<String, String> doNotMatch);
28+
29+
}

‎src/test/java/com/dtforce/spring/kubernetes/tests/DiscoveryClientTests.java‎

Lines changed: 40 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
package com.dtforce.spring.kubernetes.tests;
22

33
import com.dtforce.spring.kubernetes.KubernetesDiscoveryClient;
4-
import io.fabric8.kubernetes.api.model.Service;
5-
import io.fabric8.kubernetes.api.model.ServiceBuilder;
64
import io.fabric8.kubernetes.client.KubernetesClient;
75
import io.fabric8.kubernetes.client.server.mock.KubernetesServer;
86
import org.junit.Before;
@@ -12,6 +10,7 @@
1210
import org.springframework.cloud.client.ServiceInstance;
1311
import org.springframework.cloud.client.discovery.DiscoveryClient;
1412

13+
import javax.xml.ws.Service;
1514
import java.util.HashMap;
1615
import java.util.List;
1716
import java.util.Map;
@@ -24,6 +23,8 @@ public class DiscoveryClientTests
2423

2524
private final String serviceId = "dummy-service";
2625

26+
private final String multiPortServiceId = "multiport-service";
27+
2728
@Rule
2829
public KubernetesServer server = new KubernetesServer(true, true);
2930

@@ -43,9 +44,10 @@ public void setUp()
4344
svcLabels.put("beta.kubernetes.io/arch", "amd64");
4445
svcLabels.put("beta.kubernetes.io/os", "linux");
4546

46-
Service svc = new ServiceBuilder()
47+
kube.services().createNew()
4748
.withNewMetadata()
4849
.withName(serviceId)
50+
.withNamespace(kube.getNamespace())
4951
.withLabels(svcLabels)
5052
.and()
5153
.withNewSpec()
@@ -55,8 +57,25 @@ public void setUp()
5557
.withPort(80).withProtocol("TCP")
5658
.endPort()
5759
.and()
58-
.build();
59-
kube.services().create(svc);
60+
.done();
61+
62+
kube.services().createNew()
63+
.withNewMetadata()
64+
.withName(multiPortServiceId)
65+
.withNamespace(kube.getNamespace())
66+
.withLabels(svcLabels)
67+
.and()
68+
.withNewSpec()
69+
.withType("ClusterIP")
70+
.withClusterIP("192.168.1.120")
71+
.addNewPort()
72+
.withPort(80).withProtocol("TCP")
73+
.endPort()
74+
.addNewPort()
75+
.withPort(8080).withProtocol("TCP").withName("http")
76+
.endPort()
77+
.and()
78+
.done();
6079
}
6180

6281
@Test
@@ -69,15 +88,25 @@ public void listServices()
6988
@Test
7089
public void getInstancesForSpecifiedService()
7190
{
72-
List<ServiceInstance> services = discoveryClient.getInstances(serviceId);
73-
assertThat(services).hasAtLeastOneElementOfType(ServiceInstance.class);
74-
validateServiceInstance(services.get(0));
91+
List<ServiceInstance> instances = discoveryClient.getInstances(serviceId);
92+
assertThat(instances).hasAtLeastOneElementOfType(ServiceInstance.class);
93+
validateServiceInstance(instances.get(0));
7594
}
7695

7796
@Test
78-
public void getInstancesForNonExistentService() {
79-
List<ServiceInstance> services = discoveryClient.getInstances("noop");
80-
assertThat(services).isEmpty();
97+
public void getInstancesForNonExistentService()
98+
{
99+
List<ServiceInstance> instances = discoveryClient.getInstances("noop");
100+
assertThat(instances).isEmpty();
101+
}
102+
103+
@Test
104+
public void getHttpInstanceForMultiPortService() {
105+
List<ServiceInstance> instances = discoveryClient.getInstances(multiPortServiceId);
106+
assertThat(instances).hasSize(1);
107+
validateServiceInstance(instances.get(0));
108+
109+
assertThat(instances.get(0).getPort()).isEqualTo(8080);
81110
}
82111

83112
private void validateServiceInstance(ServiceInstance serviceInstance)

0 commit comments

Comments
 (0)