This is an automated email from the ASF dual-hosted git repository. wangyang0918 pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit f24dc6da25246fdaf278f4b2ce88892ec79d1b8b Author: Aitozi <[email protected]> AuthorDate: Mon Feb 14 23:22:25 2022 +0800 [FLINK-26112][k8s]Port getRestEndpoint method to the specific service type subclass This closes #18762. --- .../kubeclient/Fabric8FlinkKubeClient.java | 110 +-------------------- .../decorators/ExternalServiceDecorator.java | 5 +- .../kubeclient/services/ClusterIPService.java | 23 +++++ .../kubeclient/services/LoadBalancerService.java | 92 +++++++++++++++++ .../kubeclient/services/NodePortService.java | 48 +++++++++ .../kubeclient/services/ServiceType.java | 49 +++++++++ .../flink/kubernetes/utils/KubernetesUtils.java | 6 ++ .../flink/kubernetes/KubernetesClientTestBase.java | 1 + 8 files changed, 226 insertions(+), 108 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 2c8f7541ccd..023ed5df3d8 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 @@ -31,7 +31,6 @@ import org.apache.flink.kubernetes.kubeclient.resources.KubernetesPodsWatcher; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesService; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch; import org.apache.flink.kubernetes.kubeclient.services.ServiceType; -import org.apache.flink.kubernetes.utils.Constants; import org.apache.flink.kubernetes.utils.KubernetesUtils; import org.apache.flink.runtime.persistence.PossibleInconsistentStateException; import org.apache.flink.util.ExceptionUtils; @@ -42,14 +41,11 @@ import org.apache.flink.util.concurrent.FutureUtils; import io.fabric8.kubernetes.api.model.ConfigMap; import io.fabric8.kubernetes.api.model.HasMetadata; import io.fabric8.kubernetes.api.model.IntOrString; -import io.fabric8.kubernetes.api.model.LoadBalancerStatus; -import io.fabric8.kubernetes.api.model.NodeAddress; import io.fabric8.kubernetes.api.model.OwnerReference; import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder; import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.Service; import io.fabric8.kubernetes.api.model.ServiceBuilder; -import io.fabric8.kubernetes.api.model.ServicePort; import io.fabric8.kubernetes.api.model.apps.Deployment; import io.fabric8.kubernetes.client.KubernetesClientException; import io.fabric8.kubernetes.client.NamespacedKubernetesClient; @@ -184,21 +180,13 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient { return Optional.empty(); } final Service service = restService.get().getInternalResource(); - final int restPort = getRestPortFromExternalService(service); final KubernetesConfigOptions.ServiceExposedType serviceExposedType = ServiceType.classify(service); - // Return the external service.namespace directly when using ClusterIP. - if (serviceExposedType.isClusterIP()) { - return Optional.of( - new Endpoint( - ExternalServiceDecorator.getNamespacedExternalServiceName( - clusterId, namespace), - restPort)); - } - - return getRestEndPointFromService(service, restPort); + return serviceExposedType + .serviceType() + .getRestEndpoint(service, internalClient, nodePortAddressType); } @Override @@ -432,96 +420,4 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient { .setOwnerReferences( Collections.singletonList(deploymentOwnerReference))); } - - /** Get rest port from the external Service. */ - private int getRestPortFromExternalService(Service externalService) { - final List<ServicePort> servicePortCandidates = - externalService.getSpec().getPorts().stream() - .filter(x -> x.getName().equals(Constants.REST_PORT_NAME)) - .collect(Collectors.toList()); - - if (servicePortCandidates.isEmpty()) { - throw new RuntimeException( - "Failed to find port \"" - + Constants.REST_PORT_NAME - + "\" in Service \"" - + ExternalServiceDecorator.getExternalServiceName(this.clusterId) - + "\""); - } - - final ServicePort externalServicePort = servicePortCandidates.get(0); - - final KubernetesConfigOptions.ServiceExposedType externalServiceType = - KubernetesConfigOptions.ServiceExposedType.valueOf( - externalService.getSpec().getType()); - - switch (externalServiceType) { - case ClusterIP: - case LoadBalancer: - return externalServicePort.getPort(); - case NodePort: - return externalServicePort.getNodePort(); - default: - throw new RuntimeException("Unrecognized Service type: " + externalServiceType); - } - } - - private Optional<Endpoint> getRestEndPointFromService(Service service, int restPort) { - if (service.getStatus() == null) { - return Optional.empty(); - } - - LoadBalancerStatus loadBalancer = service.getStatus().getLoadBalancer(); - boolean hasExternalIP = - service.getSpec() != null - && service.getSpec().getExternalIPs() != null - && !service.getSpec().getExternalIPs().isEmpty(); - - if (loadBalancer != null) { - return getLoadBalancerRestEndpoint(loadBalancer, restPort); - } else if (hasExternalIP) { - final String address = service.getSpec().getExternalIPs().get(0); - if (address != null && !address.isEmpty()) { - return Optional.of(new Endpoint(address, restPort)); - } - } - return Optional.empty(); - } - - private Optional<Endpoint> getLoadBalancerRestEndpoint( - LoadBalancerStatus loadBalancer, int restPort) { - boolean hasIngress = - loadBalancer.getIngress() != null && !loadBalancer.getIngress().isEmpty(); - String address; - if (hasIngress) { - address = loadBalancer.getIngress().get(0).getIp(); - // Use hostname when the ip address is null - if (address == null || address.isEmpty()) { - address = loadBalancer.getIngress().get(0).getHostname(); - } - } else { - // Use node port. Node port is accessible on any node within kubernetes cluster. We'll - // only consider IPs with the configured address type. - address = - internalClient.nodes().list().getItems().stream() - .flatMap(node -> node.getStatus().getAddresses().stream()) - .filter( - nodeAddress -> - nodePortAddressType - .name() - .equals(nodeAddress.getType())) - .map(NodeAddress::getAddress) - .filter(ip -> !ip.isEmpty()) - .findAny() - .orElse(null); - if (address == null) { - LOG.warn( - "Unable to find any node ip with type [{}]. Please see [{}] config option for more details.", - nodePortAddressType, - KubernetesConfigOptions.REST_SERVICE_EXPOSED_NODE_PORT_ADDRESS_TYPE.key()); - } - } - boolean noAddress = address == null || address.isEmpty(); - return noAddress ? Optional.empty() : Optional.of(new Endpoint(address, restPort)); - } } diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java index bda69522523..d42f4513ddb 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java @@ -55,7 +55,10 @@ public class ExternalServiceDecorator extends AbstractKubernetesStepDecorator { return clusterId + Constants.FLINK_REST_SERVICE_SUFFIX; } - /** Generate namespaced name of the external rest Service. */ + /** + * Generate namespaced name of the external rest Service by cluster Id, This is used by other + * project, so do not delete it. + */ public static String getNamespacedExternalServiceName(String clusterId, String namespace) { return getExternalServiceName(clusterId) + "." + namespace; } diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ClusterIPService.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ClusterIPService.java index a6a3d6b9e84..f206756b09a 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ClusterIPService.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ClusterIPService.java @@ -19,9 +19,15 @@ package org.apache.flink.kubernetes.kubeclient.services; import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions; +import org.apache.flink.kubernetes.kubeclient.Endpoint; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters; +import org.apache.flink.kubernetes.utils.KubernetesUtils; import io.fabric8.kubernetes.api.model.Service; +import io.fabric8.kubernetes.api.model.ServicePort; +import io.fabric8.kubernetes.client.NamespacedKubernetesClient; + +import java.util.Optional; /** The service type of ClusterIP. */ public class ClusterIPService extends ServiceType { @@ -34,6 +40,23 @@ public class ClusterIPService extends ServiceType { throw new UnsupportedOperationException(); } + @Override + public Optional<Endpoint> getRestEndpoint( + Service targetService, + NamespacedKubernetesClient internalClient, + KubernetesConfigOptions.NodePortAddressType nodePortAddressType) { + int restPort = getRestPortFromExternalService(targetService); + + // Return the external service.namespace directly when using ClusterIP. + return Optional.of( + new Endpoint(KubernetesUtils.getNamespacedServiceName(targetService), restPort)); + } + + @Override + public int getRestPort(ServicePort port) { + return port.getPort(); + } + @Override public String getType() { return KubernetesConfigOptions.ServiceExposedType.ClusterIP.name(); diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/LoadBalancerService.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/LoadBalancerService.java index 57b2a36f74f..5eb4f306d51 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/LoadBalancerService.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/LoadBalancerService.java @@ -19,13 +19,23 @@ package org.apache.flink.kubernetes.kubeclient.services; import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions; +import org.apache.flink.kubernetes.kubeclient.Endpoint; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters; +import io.fabric8.kubernetes.api.model.LoadBalancerStatus; +import io.fabric8.kubernetes.api.model.NodeAddress; import io.fabric8.kubernetes.api.model.Service; +import io.fabric8.kubernetes.api.model.ServicePort; +import io.fabric8.kubernetes.client.NamespacedKubernetesClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Optional; /** The service type of LoadBalancer. */ public class LoadBalancerService extends ServiceType { + private static final Logger LOG = LoggerFactory.getLogger(LoadBalancerService.class); public static final LoadBalancerService INSTANCE = new LoadBalancerService(); @Override @@ -34,8 +44,90 @@ public class LoadBalancerService extends ServiceType { throw new UnsupportedOperationException(); } + @Override + public Optional<Endpoint> getRestEndpoint( + Service targetService, + NamespacedKubernetesClient internalClient, + KubernetesConfigOptions.NodePortAddressType nodePortAddressType) { + int restPort = getRestPortFromExternalService(targetService); + return getRestEndPointFromService( + internalClient, nodePortAddressType, targetService, restPort); + } + + @Override + public int getRestPort(ServicePort port) { + return port.getPort(); + } + @Override public String getType() { return KubernetesConfigOptions.ServiceExposedType.LoadBalancer.name(); } + + private Optional<Endpoint> getRestEndPointFromService( + NamespacedKubernetesClient internalClient, + KubernetesConfigOptions.NodePortAddressType nodePortAddressType, + Service service, + int restPort) { + if (service.getStatus() == null) { + return Optional.empty(); + } + + LoadBalancerStatus loadBalancer = service.getStatus().getLoadBalancer(); + boolean hasExternalIP = + service.getSpec() != null + && service.getSpec().getExternalIPs() != null + && !service.getSpec().getExternalIPs().isEmpty(); + + if (loadBalancer != null) { + return getLoadBalancerRestEndpoint( + internalClient, nodePortAddressType, loadBalancer, restPort); + } else if (hasExternalIP) { + final String address = service.getSpec().getExternalIPs().get(0); + if (address != null && !address.isEmpty()) { + return Optional.of(new Endpoint(address, restPort)); + } + } + return Optional.empty(); + } + + private Optional<Endpoint> getLoadBalancerRestEndpoint( + NamespacedKubernetesClient internalClient, + KubernetesConfigOptions.NodePortAddressType nodePortAddressType, + LoadBalancerStatus loadBalancer, + int restPort) { + boolean hasIngress = + loadBalancer.getIngress() != null && !loadBalancer.getIngress().isEmpty(); + String address; + if (hasIngress) { + address = loadBalancer.getIngress().get(0).getIp(); + // Use hostname when the ip address is null + if (address == null || address.isEmpty()) { + address = loadBalancer.getIngress().get(0).getHostname(); + } + } else { + // Use node port. Node port is accessible on any node within kubernetes cluster. We'll + // only consider IPs with the configured address type. + address = + internalClient.nodes().list().getItems().stream() + .flatMap(node -> node.getStatus().getAddresses().stream()) + .filter( + nodeAddress -> + nodePortAddressType + .name() + .equals(nodeAddress.getType())) + .map(NodeAddress::getAddress) + .filter(ip -> !ip.isEmpty()) + .findAny() + .orElse(null); + if (address == null) { + LOG.warn( + "Unable to find any node ip with type [{}]. Please see [{}] config option for more details.", + nodePortAddressType, + KubernetesConfigOptions.REST_SERVICE_EXPOSED_NODE_PORT_ADDRESS_TYPE.key()); + } + } + boolean noAddress = address == null || address.isEmpty(); + return noAddress ? Optional.empty() : Optional.of(new Endpoint(address, restPort)); + } } diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/NodePortService.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/NodePortService.java index 9fcf1a7e43e..326eb663ea1 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/NodePortService.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/NodePortService.java @@ -19,13 +19,23 @@ package org.apache.flink.kubernetes.kubeclient.services; import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions; +import org.apache.flink.kubernetes.kubeclient.Endpoint; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters; +import org.apache.flink.util.StringUtils; +import io.fabric8.kubernetes.api.model.NodeAddress; import io.fabric8.kubernetes.api.model.Service; +import io.fabric8.kubernetes.api.model.ServicePort; +import io.fabric8.kubernetes.client.NamespacedKubernetesClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Optional; /** The service type of NodePort. */ public class NodePortService extends ServiceType { + private static final Logger LOG = LoggerFactory.getLogger(NodePortService.class); public static final NodePortService INSTANCE = new NodePortService(); private NodePortService() {} @@ -36,6 +46,44 @@ public class NodePortService extends ServiceType { throw new UnsupportedOperationException(); } + @Override + public Optional<Endpoint> getRestEndpoint( + Service targetService, + NamespacedKubernetesClient internalClient, + KubernetesConfigOptions.NodePortAddressType nodePortAddressType) { + if (targetService.getStatus() == null) { + return Optional.empty(); + } + + String address; + // Node port is accessible on any node within kubernetes cluster. We'll + // only consider IPs with the configured address type. + address = + internalClient.nodes().list().getItems().stream() + .flatMap(node -> node.getStatus().getAddresses().stream()) + .filter( + nodeAddress -> + nodePortAddressType.name().equals(nodeAddress.getType())) + .map(NodeAddress::getAddress) + .filter(ip -> !ip.isEmpty()) + .findAny() + .orElse(null); + if (address == null) { + LOG.warn( + "Unable to find any node ip with type [{}]. Please see [{}] config option for more details.", + nodePortAddressType, + KubernetesConfigOptions.REST_SERVICE_EXPOSED_NODE_PORT_ADDRESS_TYPE.key()); + } + return StringUtils.isNullOrWhitespaceOnly(address) + ? Optional.empty() + : Optional.of(new Endpoint(address, getRestPortFromExternalService(targetService))); + } + + @Override + public int getRestPort(ServicePort port) { + return port.getNodePort(); + } + @Override public String getType() { return KubernetesConfigOptions.ServiceExposedType.NodePort.name(); diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ServiceType.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ServiceType.java index 5bfe0470c00..7952796b8d6 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ServiceType.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/services/ServiceType.java @@ -19,12 +19,19 @@ package org.apache.flink.kubernetes.kubeclient.services; import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions; +import org.apache.flink.kubernetes.kubeclient.Endpoint; import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters; import org.apache.flink.kubernetes.utils.Constants; import io.fabric8.kubernetes.api.model.Service; import io.fabric8.kubernetes.api.model.ServiceBuilder; +import io.fabric8.kubernetes.api.model.ServicePort; +import io.fabric8.kubernetes.client.NamespacedKubernetesClient; + +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; /** An abstract class represents the service type that flink supported. */ public abstract class ServiceType { @@ -73,6 +80,27 @@ public abstract class ServiceType { public abstract Service buildUpInternalService( KubernetesJobManagerParameters kubernetesJobManagerParameters); + /** + * Gets the rest endpoint from the kubernetes service. + * + * @param targetService the target service to retrieve from. + * @param internalClient the client to interact with api server. + * @param nodePortAddressType the target address type of the node. + * @return the rest endpoint. + */ + public abstract Optional<Endpoint> getRestEndpoint( + Service targetService, + NamespacedKubernetesClient internalClient, + KubernetesConfigOptions.NodePortAddressType nodePortAddressType); + + /** + * Gets the rest port from the service port. + * + * @param port the target service port. + * @return the rest port. + */ + public abstract int getRestPort(ServicePort port); + /** * Gets the type of the target kubernetes service. * @@ -80,6 +108,27 @@ public abstract class ServiceType { */ public abstract String getType(); + /** Get rest port from the external Service. */ + public int getRestPortFromExternalService(Service externalService) { + final List<ServicePort> servicePortCandidates = + externalService.getSpec().getPorts().stream() + .filter(x -> x.getName().equals(Constants.REST_PORT_NAME)) + .collect(Collectors.toList()); + + if (servicePortCandidates.isEmpty()) { + throw new RuntimeException( + "Failed to find port \"" + + Constants.REST_PORT_NAME + + "\" in Service \"" + + externalService.getMetadata().getName() + + "\""); + } + + final ServicePort externalServicePort = servicePortCandidates.get(0); + + return getRestPort(externalServicePort); + } + // Helper method public static KubernetesConfigOptions.ServiceExposedType classify(Service service) { KubernetesConfigOptions.ServiceExposedType type = diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java index f8afbafde0a..f98fc307b5e 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java @@ -61,6 +61,7 @@ import io.fabric8.kubernetes.api.model.KubernetesResource; import io.fabric8.kubernetes.api.model.Quantity; import io.fabric8.kubernetes.api.model.ResourceRequirements; import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder; +import io.fabric8.kubernetes.api.model.Service; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -616,5 +617,10 @@ public class KubernetesUtils { return key.substring(LEADER_PREFIX.length()); } + /** Generate namespaced name of the service. */ + public static String getNamespacedServiceName(Service service) { + return service.getMetadata().getName() + "." + service.getMetadata().getNamespace(); + } + private KubernetesUtils() {} } diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java index 26dc459e685..80ed0504c80 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java @@ -211,6 +211,7 @@ public class KubernetesClientTestBase extends KubernetesTestBase { new ServiceBuilder() .editOrNewMetadata() .withName(ExternalServiceDecorator.getExternalServiceName(CLUSTER_ID)) + .withNamespace(NAMESPACE) .endMetadata() .editOrNewSpec() .withType(serviceExposedType.name())
