You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/12/17 09:19:33 UTC
[flink] branch release-1.10 updated: [FLINK-15251][kubernetes] Use
hostname to create end point when ip in load balancer is null
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.10 by this push:
new 2b085ff [FLINK-15251][kubernetes] Use hostname to create end point when ip in load balancer is null
2b085ff is described below
commit 2b085ff094b0bdbc245951c0f57fbfd66f48ddf6
Author: wangyang0918 <da...@alibaba-inc.com>
AuthorDate: Mon Dec 16 20:02:46 2019 +0800
[FLINK-15251][kubernetes] Use hostname to create end point when ip in load balancer is null
---
.../kubeclient/Fabric8FlinkKubeClient.java | 8 +++--
.../flink/kubernetes/KubernetesTestBase.java | 26 +++++++++--------
.../kubernetes/kubeclient/Fabric8ClientTest.java | 34 ++++++++++++++++++++++
3 files changed, 54 insertions(+), 14 deletions(-)
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
index 3a3a6c6..74df4c4 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
@@ -191,19 +191,23 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient {
}
Service service = restService.getInternalResource();
- String address;
+ String address = null;
if (service.getStatus() != null && (service.getStatus().getLoadBalancer() != null ||
service.getStatus().getLoadBalancer().getIngress() != null)) {
if (service.getStatus().getLoadBalancer().getIngress().size() > 0) {
address = service.getStatus().getLoadBalancer().getIngress().get(0).getIp();
+ if (address == null || address.isEmpty()) {
+ address = service.getStatus().getLoadBalancer().getIngress().get(0).getHostname();
+ }
} else {
address = this.internalClient.getMasterUrl().getHost();
restPort = getServiceNodePort(service, RestOptions.PORT);
}
} else if (service.getSpec().getExternalIPs() != null && service.getSpec().getExternalIPs().size() > 0) {
address = service.getSpec().getExternalIPs().get(0);
- } else {
+ }
+ if (address == null || address.isEmpty()) {
return null;
}
return new Endpoint(address, restPort);
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
index b577e58..551f11f 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
@@ -46,6 +46,8 @@ import org.junit.Before;
import org.junit.Rule;
import org.junit.rules.TemporaryFolder;
+import javax.annotation.Nullable;
+
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
@@ -76,6 +78,8 @@ public class KubernetesTestBase extends TestLogger {
protected static final String MOCK_SERVICE_ID = "mock-uuid-of-service";
+ protected static final String MOCK_SERVICE_HOST_NAME = "mock-host-name-of-service";
+
protected static final String MOCK_SERVICE_IP = "192.168.0.1";
@Before
@@ -96,8 +100,8 @@ public class KubernetesTestBase extends TestLogger {
// Set mock requests.
mockInternalServiceActionWatch();
- mockRestServiceActionWatcher();
- mockGetRestService();
+ mockRestServiceActionWatcher(CLUSTER_ID);
+ mockGetRestService(CLUSTER_ID, MOCK_SERVICE_HOST_NAME, MOCK_SERVICE_IP);
}
protected FlinkKubeClient getFabric8FlinkKubeClient(){
@@ -112,8 +116,8 @@ public class KubernetesTestBase extends TestLogger {
return server.getClient().inNamespace(NAMESPACE);
}
- private void mockRestServiceActionWatcher() {
- String serviceName = CLUSTER_ID + Constants.FLINK_REST_SERVICE_SUFFIX;
+ protected void mockRestServiceActionWatcher(String clusterId) {
+ String serviceName = clusterId + Constants.FLINK_REST_SERVICE_SUFFIX;
String path = String.format("/api/v1/namespaces/%s/services?fieldSelector=metadata.name%%3D%s&watch=true",
NAMESPACE, serviceName);
@@ -122,7 +126,7 @@ public class KubernetesTestBase extends TestLogger {
.andUpgradeToWebSocket()
.open()
.waitFor(1000)
- .andEmit(new WatchEvent(getMockRestService(), "ADDED"))
+ .andEmit(new WatchEvent(getMockRestService(MOCK_SERVICE_HOST_NAME, MOCK_SERVICE_IP), "ADDED"))
.done()
.once();
}
@@ -141,17 +145,17 @@ public class KubernetesTestBase extends TestLogger {
.once();
}
- private void mockGetRestService() {
- String serviceName = CLUSTER_ID + Constants.FLINK_REST_SERVICE_SUFFIX;
+ protected void mockGetRestService(String clusterId, @Nullable String hostname, @Nullable String ip) {
+ String serviceName = clusterId + Constants.FLINK_REST_SERVICE_SUFFIX;
String path = String.format("/api/v1/namespaces/%s/services/%s", NAMESPACE, serviceName);
server.expect()
.withPath(path)
- .andReturn(200, getMockRestService())
+ .andReturn(200, getMockRestService(hostname, ip))
.always();
}
- private Service getMockRestService() {
+ private Service getMockRestService(@Nullable String hostname, @Nullable String ip) {
List<Decorator<Service, KubernetesService>> restServiceDecorators = new ArrayList<>();
restServiceDecorators.add(new InitializerDecorator<>(CLUSTER_ID + Constants.FLINK_REST_SERVICE_SUFFIX));
String exposedType = FLINK_CONFIG.getString(KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE);
@@ -165,10 +169,9 @@ public class KubernetesTestBase extends TestLogger {
}
Service service = kubernetesService.getInternalResource();
- String mockServiceHostName = "mock-host-name-of-service";
service.setStatus(new ServiceStatusBuilder()
.withLoadBalancer(new LoadBalancerStatus(Collections.singletonList(
- new LoadBalancerIngress(mockServiceHostName, MOCK_SERVICE_IP)))).build());
+ new LoadBalancerIngress(hostname, ip)))).build());
return service;
}
@@ -192,5 +195,4 @@ public class KubernetesTestBase extends TestLogger {
labels.put(Constants.LABEL_APP_KEY, CLUSTER_ID);
return labels;
}
-
}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java
index 08ebedf..6d54617 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8ClientTest.java
@@ -47,6 +47,8 @@ import java.util.stream.Collectors;
import static org.apache.flink.configuration.GlobalConfiguration.FLINK_CONF_FILENAME;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
/**
@@ -245,4 +247,36 @@ public class Fabric8ClientTest extends KubernetesTestBase {
flinkKubeClient.stopPod(podName);
assertEquals(0, kubeClient.pods().list().getItems().size());
}
+
+ @Test
+ public void testServiceLoadBalancerWithNoIP() throws Exception {
+ final String hostName = "test-host-name";
+ final Endpoint endpoint = getRestEndpoint(hostName, "");
+ assertEquals(hostName, endpoint.getAddress());
+ assertEquals(8081, endpoint.getPort());
+ }
+
+ @Test
+ public void testServiceLoadBalancerEmptyHostAndIP() throws Exception {
+ final Endpoint endpoint1 = getRestEndpoint("", "");
+ assertNull(endpoint1);
+
+ final Endpoint endpoint2 = getRestEndpoint(null, null);
+ assertNull(endpoint2);
+ }
+
+ private Endpoint getRestEndpoint(String hostName, String ip) throws Exception {
+ final String clusterId = "flink-on-k8s-cluster-test";
+ mockRestServiceActionWatcher(clusterId);
+ mockGetRestService(clusterId, hostName, ip);
+
+ flinkKubeClient.createRestService(clusterId).get();
+
+ final Service services = kubeClient.services()
+ .withName(clusterId + Constants.FLINK_REST_SERVICE_SUFFIX)
+ .get();
+ assertNotNull(services);
+
+ return flinkKubeClient.getRestEndpoint(clusterId);
+ }
}