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 ab3eb40d920fa609f49164a0bbb5fcbb3004a808 Author: Aitozi <[email protected]> AuthorDate: Mon Feb 14 15:56:14 2022 +0800 [Flink-26112][k8s]Refactor the FlinkKubeClient interface to deal with service name directly --- .../kubernetes/KubernetesClusterDescriptor.java | 5 +-- .../KubernetesResourceManagerDriver.java | 18 +++++------ .../flink/kubernetes/cli/KubernetesSessionCli.java | 4 +-- .../kubeclient/Fabric8FlinkKubeClient.java | 36 +++------------------- .../kubernetes/kubeclient/FlinkKubeClient.java | 15 +++------ .../kubeclient/resources/KubernetesService.java | 6 ---- .../kubeclient/Fabric8FlinkKubeClientTest.java | 15 ++++----- .../kubeclient/TestingFlinkKubeClient.java | 8 ++--- 8 files changed, 31 insertions(+), 76 deletions(-) diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java index 319ba3a48a4..8a2b0d4d725 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java @@ -42,9 +42,9 @@ import org.apache.flink.kubernetes.kubeclient.Endpoint; import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient; import org.apache.flink.kubernetes.kubeclient.FlinkPod; import org.apache.flink.kubernetes.kubeclient.KubernetesJobManagerSpecification; +import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator; import org.apache.flink.kubernetes.kubeclient.factory.KubernetesJobManagerFactory; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters; -import org.apache.flink.kubernetes.kubeclient.resources.KubernetesService; import org.apache.flink.kubernetes.utils.Constants; import org.apache.flink.kubernetes.utils.KubernetesUtils; import org.apache.flink.runtime.entrypoint.ClusterEntrypoint; @@ -177,7 +177,8 @@ public class KubernetesClusterDescriptor implements ClusterDescriptor<String> { final ClusterSpecification clusterSpecification, final ApplicationConfiguration applicationConfiguration) throws ClusterDeploymentException { - if (client.getService(KubernetesService.ServiceType.REST_SERVICE, clusterId).isPresent()) { + if (client.getService(ExternalServiceDecorator.getExternalServiceName(clusterId)) + .isPresent()) { throw new ClusterDeploymentException( "The Flink cluster " + clusterId + " already exists."); } diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java index 12105504e48..8023e70af62 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java @@ -27,10 +27,11 @@ import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions; import org.apache.flink.kubernetes.configuration.KubernetesResourceManagerDriverConfiguration; import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient; import org.apache.flink.kubernetes.kubeclient.FlinkPod; +import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator; +import org.apache.flink.kubernetes.kubeclient.decorators.InternalServiceDecorator; import org.apache.flink.kubernetes.kubeclient.factory.KubernetesTaskManagerFactory; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesTaskManagerParameters; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesPod; -import org.apache.flink.kubernetes.kubeclient.resources.KubernetesService; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesTooOldResourceVersionException; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch; import org.apache.flink.kubernetes.utils.Constants; @@ -257,25 +258,22 @@ public class KubernetesResourceManagerDriver ResourceManagerUtils.parseRestBindPortFromWebInterfaceUrl(webInterfaceUrl); Preconditions.checkArgument( restPort > 0, "Failed to parse rest port from " + webInterfaceUrl); + final String restServiceName = ExternalServiceDecorator.getExternalServiceName(clusterId); flinkKubeClient - .updateServiceTargetPort( - KubernetesService.ServiceType.REST_SERVICE, - clusterId, - Constants.REST_PORT_NAME, - restPort) + .updateServiceTargetPort(restServiceName, Constants.REST_PORT_NAME, restPort) .get(); if (!HighAvailabilityMode.isHighAvailabilityModeActivated(flinkConfig)) { + final String internalServiceName = + InternalServiceDecorator.getInternalServiceName(clusterId); flinkKubeClient .updateServiceTargetPort( - KubernetesService.ServiceType.INTERNAL_SERVICE, - clusterId, + internalServiceName, Constants.BLOB_SERVER_PORT_NAME, Integer.parseInt(flinkConfig.getString(BlobServerOptions.PORT))) .get(); flinkKubeClient .updateServiceTargetPort( - KubernetesService.ServiceType.INTERNAL_SERVICE, - clusterId, + internalServiceName, Constants.JOB_MANAGER_RPC_PORT_NAME, flinkConfig.getInteger(JobManagerOptions.PORT)) .get(); diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java index b0e8692575d..36a1242b2a5 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java @@ -36,7 +36,7 @@ import org.apache.flink.configuration.UnmodifiableConfiguration; import org.apache.flink.kubernetes.executors.KubernetesSessionClusterExecutor; import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient; import org.apache.flink.kubernetes.kubeclient.FlinkKubeClientFactory; -import org.apache.flink.kubernetes.kubeclient.resources.KubernetesService; +import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator; import org.apache.flink.runtime.security.SecurityUtils; import org.apache.flink.util.FlinkException; @@ -109,7 +109,7 @@ public class KubernetesSessionCli { // Retrieve or create a session cluster. if (clusterId != null && kubeClient - .getService(KubernetesService.ServiceType.REST_SERVICE, clusterId) + .getService(ExternalServiceDecorator.getExternalServiceName(clusterId)) .isPresent()) { clusterClient = kubernetesClusterDescriptor.retrieve(clusterId).getClusterClient(); } else { 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 30dcccbe92a..2c8f7541ccd 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 @@ -22,7 +22,6 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions; import org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration; import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator; -import org.apache.flink.kubernetes.kubeclient.decorators.InternalServiceDecorator; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMapSharedInformer; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesException; @@ -180,7 +179,7 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient { @Override public Optional<Endpoint> getRestEndpoint(String clusterId) { Optional<KubernetesService> restService = - getService(KubernetesService.ServiceType.REST_SERVICE, clusterId); + getService(ExternalServiceDecorator.getExternalServiceName(clusterId)); if (!restService.isPresent()) { return Optional.empty(); } @@ -224,9 +223,7 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient { } @Override - public Optional<KubernetesService> getService( - KubernetesService.ServiceType serviceType, String clusterId) { - final String serviceName = getServiceName(serviceType, clusterId); + public Optional<KubernetesService> getService(String serviceName) { final Service service = this.internalClient.services().withName(serviceName).fromServer().get(); if (service == null) { @@ -388,14 +385,11 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient { @Override public CompletableFuture<Void> updateServiceTargetPort( - KubernetesService.ServiceType serviceType, - String clusterId, - String portName, - int targetPort) { + String serviceName, String portName, int targetPort) { LOG.debug("Update {} target port to {}", portName, targetPort); return CompletableFuture.runAsync( () -> - getService(serviceType, clusterId) + getService(serviceName) .ifPresent( service -> { final Service updatedService = @@ -416,32 +410,12 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient { .build(); this.internalClient .services() - .withName( - getServiceName(serviceType, clusterId)) + .withName(serviceName) .replace(updatedService); }), kubeClientExecutorService); } - /** - * Get the Kubernetes service name. - * - * @param serviceType The service type - * @param clusterId The cluster id - * @return Return the Kubernetes service name if the service type is known. - */ - private String getServiceName(KubernetesService.ServiceType serviceType, String clusterId) { - switch (serviceType) { - case REST_SERVICE: - return ExternalServiceDecorator.getExternalServiceName(clusterId); - case INTERNAL_SERVICE: - return InternalServiceDecorator.getInternalServiceName(clusterId); - default: - throw new IllegalArgumentException( - "Unrecognized service type: " + serviceType.name()); - } - } - private void setOwnerReference(Deployment deployment, List<HasMetadata> resources) { final OwnerReference deploymentOwnerReference = new OwnerReferenceBuilder() diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java index 47038244a3f..64b45385b05 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java @@ -76,12 +76,10 @@ public interface FlinkKubeClient extends AutoCloseable { /** * Get the kubernetes service of the given flink clusterId. * - * @param serviceType Internal/Rest - * @param clusterId cluster id - * @return Return the optional rest service of the specified cluster id. + * @param serviceName the name of the service + * @return Return the optional kubernetes service of the specified name. */ - Optional<KubernetesService> getService( - KubernetesService.ServiceType serviceType, String clusterId); + Optional<KubernetesService> getService(String serviceName); /** * Get the rest endpoint for access outside cluster. @@ -207,16 +205,13 @@ public interface FlinkKubeClient extends AutoCloseable { /** * Update the target ports of the given Kubernetes service. * - * @param serviceType The service type which needs to be updated + * @param serviceName The name of the service which needs to be updated * @param portName The port name which needs to be updated * @param targetPort The updated target port * @return Return the update service target port future */ CompletableFuture<Void> updateServiceTargetPort( - KubernetesService.ServiceType serviceType, - String clusterId, - String portName, - int targetPort); + String serviceName, String portName, int targetPort); /** Callback handler for kubernetes resources. */ interface WatchCallbackHandler<T> { diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesService.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesService.java index 86d28e80031..2af825aef75 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesService.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesService.java @@ -26,10 +26,4 @@ public class KubernetesService extends KubernetesResource<Service> { public KubernetesService(Service internalResource) { super(internalResource); } - - /** The flink service type. */ - public enum ServiceType { - REST_SERVICE, - INTERNAL_SERVICE, - } } diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java index 412e42efa27..de95997a996 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java @@ -30,11 +30,11 @@ import org.apache.flink.kubernetes.configuration.KubernetesConfigOptionsInternal import org.apache.flink.kubernetes.configuration.KubernetesDeploymentTarget; import org.apache.flink.kubernetes.entrypoint.KubernetesSessionClusterEntrypoint; import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator; +import org.apache.flink.kubernetes.kubeclient.decorators.InternalServiceDecorator; import org.apache.flink.kubernetes.kubeclient.factory.KubernetesJobManagerFactory; import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap; import org.apache.flink.kubernetes.kubeclient.resources.KubernetesPod; -import org.apache.flink.kubernetes.kubeclient.resources.KubernetesService; import org.apache.flink.kubernetes.utils.Constants; import org.apache.flink.runtime.persistence.PossibleInconsistentStateException; import org.apache.flink.util.concurrent.ExecutorThreadFactory; @@ -172,12 +172,10 @@ public class Fabric8FlinkKubeClientTest extends KubernetesClientTestBase { this.flinkKubeClient.createJobManagerComponent(this.kubernetesJobManagerSpecification); final int expectedRestPort = 9081; + final String restServiceName = ExternalServiceDecorator.getExternalServiceName(CLUSTER_ID); flinkKubeClient .updateServiceTargetPort( - KubernetesService.ServiceType.REST_SERVICE, - CLUSTER_ID, - Constants.REST_PORT_NAME, - expectedRestPort) + restServiceName, Constants.REST_PORT_NAME, expectedRestPort) .get(); final int updatedRestPort = getServiceTargetPort( @@ -190,12 +188,11 @@ public class Fabric8FlinkKubeClientTest extends KubernetesClientTestBase { this.flinkKubeClient.createJobManagerComponent(this.kubernetesJobManagerSpecification); final int expectedBlobPort = 9082; + final String internalServiceName = + InternalServiceDecorator.getInternalServiceName(CLUSTER_ID); flinkKubeClient .updateServiceTargetPort( - KubernetesService.ServiceType.INTERNAL_SERVICE, - CLUSTER_ID, - Constants.BLOB_SERVER_PORT_NAME, - expectedBlobPort) + internalServiceName, Constants.BLOB_SERVER_PORT_NAME, expectedBlobPort) .get(); final int updatedBlobPort = getServiceTargetPort(CLUSTER_ID, Constants.BLOB_SERVER_PORT_NAME); diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java index 639e6f3fa41..be67ffa7eea 100644 --- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java +++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java @@ -138,8 +138,7 @@ public class TestingFlinkKubeClient implements FlinkKubeClient { } @Override - public Optional<KubernetesService> getService( - KubernetesService.ServiceType serviceType, String clusterId) { + public Optional<KubernetesService> getService(String serviceName) { throw new UnsupportedOperationException(); } @@ -211,10 +210,7 @@ public class TestingFlinkKubeClient implements FlinkKubeClient { @Override public CompletableFuture<Void> updateServiceTargetPort( - KubernetesService.ServiceType serviceType, - String clusterId, - String portName, - int targetPort) { + String serviceName, String portName, int targetPort) { throw new UnsupportedOperationException(); }
