| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.streampark.flink.client.impl |
| |
| import org.apache.streampark.common.enums.FlinkExecutionMode |
| import org.apache.streampark.common.util.{Logger, Utils} |
| import org.apache.streampark.flink.client.`trait`.KubernetesNativeClientTrait |
| import org.apache.streampark.flink.client.bean._ |
| import org.apache.streampark.flink.client.tool.FlinkSessionSubmitHelper |
| import org.apache.streampark.flink.core.FlinkKubernetesClient |
| import org.apache.streampark.flink.kubernetes.KubernetesRetriever |
| import org.apache.streampark.flink.kubernetes.enums.FlinkK8sExecuteModeEnum |
| import org.apache.streampark.flink.kubernetes.model.ClusterKey |
| |
| import io.fabric8.kubernetes.api.model.{Config => _} |
| import org.apache.commons.lang3.StringUtils |
| import org.apache.flink.client.program.{ClusterClient, PackagedProgram} |
| import org.apache.flink.configuration._ |
| import org.apache.flink.kubernetes.KubernetesClusterDescriptor |
| import org.apache.flink.kubernetes.configuration.{KubernetesConfigOptions, KubernetesDeploymentTarget} |
| import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.ServiceExposedType |
| import org.apache.flink.kubernetes.kubeclient.{FlinkKubeClient, FlinkKubeClientFactory} |
| |
| import scala.collection.convert.ImplicitConversions._ |
| import scala.language.postfixOps |
| import scala.util.{Failure, Success, Try} |
| |
| /** |
| * Kubernetes native session mode submit. |
| * @deprecated |
| * Please use [[KubernetesSessionClientV2]] instead. |
| */ |
| @Deprecated |
| object KubernetesNativeSessionClient extends KubernetesNativeClientTrait with Logger { |
| |
| @throws[Exception] |
| override def doSubmit( |
| submitRequest: SubmitRequest, |
| flinkConfig: Configuration): SubmitResponse = { |
| // require parameters |
| require( |
| StringUtils.isNotBlank(submitRequest.k8sSubmitParam.clusterId), |
| s"[flink-submit] submit flink job failed, clusterId is null, mode=${flinkConfig.get(DeploymentOptions.TARGET)}" |
| ) |
| super.trySubmit(submitRequest, flinkConfig)(jobGraphSubmit, restApiSubmit) |
| } |
| |
| /** Submit flink session job via rest api. */ |
| @throws[Exception] |
| def restApiSubmit(submitRequest: SubmitRequest, flinkConfig: Configuration): SubmitResponse = { |
| Try { |
| // get jm rest url of flink session cluster |
| val clusterKey = ClusterKey( |
| FlinkK8sExecuteModeEnum.SESSION, |
| submitRequest.k8sSubmitParam.kubernetesNamespace, |
| submitRequest.k8sSubmitParam.clusterId) |
| val jmRestUrl = KubernetesRetriever |
| .retrieveFlinkRestUrl(clusterKey) |
| .getOrElse(throw new Exception( |
| s"[flink-submit] retrieve flink session rest url failed, clusterKey=$clusterKey")) |
| // submit job via rest api |
| val jobId = |
| FlinkSessionSubmitHelper.submitViaRestApi(jmRestUrl, submitRequest.userJarFile, flinkConfig) |
| SubmitResponse(clusterKey.clusterId, flinkConfig.toMap, jobId, jmRestUrl) |
| } match { |
| case Success(s) => s |
| case Failure(e) => |
| logError(s"submit flink job fail in ${submitRequest.executionMode} mode") |
| throw e |
| } |
| } |
| |
| /** Submit flink session job with building JobGraph via ClusterClient api. */ |
| @throws[Exception] |
| def jobGraphSubmit(submitRequest: SubmitRequest, flinkConfig: Configuration): SubmitResponse = { |
| // retrieve k8s cluster and submit flink job on session mode |
| var clusterDescriptor: KubernetesClusterDescriptor = null |
| var packageProgram: PackagedProgram = null |
| var client: ClusterClient[String] = null |
| |
| try { |
| clusterDescriptor = getK8sClusterDescriptor(flinkConfig) |
| // build JobGraph |
| val programJobGraph = super.getJobGraph(submitRequest, flinkConfig) |
| packageProgram = programJobGraph._1 |
| val jobGraph = programJobGraph._2 |
| // retrieve client and submit JobGraph |
| client = clusterDescriptor |
| .retrieve(flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID)) |
| .getClusterClient |
| val submitResult = client.submitJob(jobGraph) |
| val jobId = submitResult.get().toString |
| val result = |
| SubmitResponse(client.getClusterId, flinkConfig.toMap, jobId, client.getWebInterfaceURL) |
| logInfo( |
| s"[flink-submit] flink job has been submitted. ${flinkConfIdentifierInfo(flinkConfig)}, jobId: $jobId") |
| result |
| } catch { |
| case e: Exception => |
| logError(s"submit flink job fail in ${submitRequest.executionMode} mode") |
| e.printStackTrace() |
| throw e |
| } finally { |
| if (submitRequest.safePackageProgram) { |
| Utils.close(packageProgram) |
| } |
| Utils.close(clusterDescriptor, client) |
| } |
| } |
| |
| override def doCancel( |
| cancelRequest: CancelRequest, |
| flinkConfig: Configuration): CancelResponse = { |
| flinkConfig.safeSet( |
| DeploymentOptions.TARGET, |
| FlinkExecutionMode.KUBERNETES_NATIVE_SESSION.getName) |
| super.doCancel(cancelRequest, flinkConfig) |
| } |
| |
| def deploy(deployRequest: DeployRequest): DeployResponse = { |
| logInfo( |
| s""" |
| |--------------------------------------- kubernetes cluster start --------------------------------------- |
| | userFlinkHome : ${deployRequest.flinkVersion.flinkHome} |
| | flinkVersion : ${deployRequest.flinkVersion.version} |
| | execMode : ${deployRequest.executionMode.name()} |
| | clusterId : ${deployRequest.clusterId} |
| | namespace : ${deployRequest.k8sDeployParam.kubernetesNamespace} |
| | exposedType : ${deployRequest.k8sDeployParam.flinkRestExposedType} |
| | serviceAccount : ${deployRequest.k8sDeployParam.serviceAccount} |
| | flinkImage : ${deployRequest.k8sDeployParam.flinkImage} |
| | properties : ${deployRequest.properties.mkString(",")} |
| |-------------------------------------------------------------------------------------------------------- |
| |""".stripMargin) |
| var clusterDescriptor: KubernetesClusterDescriptor = null |
| var client: ClusterClient[String] = null |
| var kubeClient: FlinkKubeClient = null |
| try { |
| val flinkConfig = |
| extractConfiguration(deployRequest.flinkVersion.flinkHome, deployRequest.properties) |
| flinkConfig |
| .safeSet(DeploymentOptions.TARGET, KubernetesDeploymentTarget.SESSION.getName) |
| .safeSet( |
| KubernetesConfigOptions.NAMESPACE, |
| deployRequest.k8sDeployParam.kubernetesNamespace) |
| .safeSet( |
| KubernetesConfigOptions.KUBERNETES_SERVICE_ACCOUNT, |
| deployRequest.k8sDeployParam.serviceAccount) |
| .safeSet( |
| KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE, |
| ServiceExposedType.valueOf(deployRequest.k8sDeployParam.flinkRestExposedType.getName)) |
| .safeSet(KubernetesConfigOptions.CLUSTER_ID, deployRequest.clusterId) |
| .safeSet(KubernetesConfigOptions.CONTAINER_IMAGE, deployRequest.k8sDeployParam.flinkImage) |
| .safeSet( |
| KubernetesConfigOptions.KUBE_CONFIG_FILE, |
| getDefaultKubernetesConf(deployRequest.k8sDeployParam.kubeConf)) |
| .safeSet( |
| DeploymentOptionsInternal.CONF_DIR, |
| s"${deployRequest.flinkVersion.flinkHome}/conf") |
| |
| val kubernetesClusterDescriptor = getK8sClusterDescriptorAndSpecification(flinkConfig) |
| clusterDescriptor = kubernetesClusterDescriptor._1 |
| kubeClient = FlinkKubeClientFactory.getInstance.fromConfiguration(flinkConfig, "client") |
| val kubeClientWrapper = new FlinkKubernetesClient(kubeClient) |
| |
| if ( |
| deployRequest.clusterId != null && kubeClientWrapper |
| .getService(deployRequest.clusterId) |
| .isPresent |
| ) { |
| client = clusterDescriptor.retrieve(deployRequest.clusterId).getClusterClient |
| } else { |
| client = |
| clusterDescriptor.deploySessionCluster(kubernetesClusterDescriptor._2).getClusterClient |
| } |
| if (client.getWebInterfaceURL != null) { |
| DeployResponse(client.getWebInterfaceURL, client.getClusterId) |
| } else { |
| null |
| } |
| } catch { |
| case e: Exception => |
| logError(s"start flink session fail in ${deployRequest.executionMode} mode") |
| e.printStackTrace() |
| throw e |
| } finally { |
| Utils.close(client, clusterDescriptor, kubeClient) |
| } |
| } |
| |
| def shutdown(shutDownRequest: ShutDownRequest): ShutDownResponse = { |
| var kubeClient: FlinkKubeClient = null |
| try { |
| val flinkConfig = getFlinkDefaultConfiguration(shutDownRequest.flinkVersion.flinkHome) |
| shutDownRequest.properties.foreach( |
| m => |
| m._2 match { |
| case v if v != null => flinkConfig.setString(m._1, m._2.toString) |
| case _ => |
| }) |
| flinkConfig |
| .safeSet(DeploymentOptions.TARGET, KubernetesDeploymentTarget.SESSION.getName) |
| .safeSet( |
| KubernetesConfigOptions.NAMESPACE, |
| shutDownRequest.kubernetesDeployParam.kubernetesNamespace) |
| .safeSet( |
| KubernetesConfigOptions.KUBERNETES_SERVICE_ACCOUNT, |
| shutDownRequest.kubernetesDeployParam.serviceAccount) |
| .safeSet( |
| KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE, |
| ServiceExposedType.valueOf( |
| shutDownRequest.kubernetesDeployParam.flinkRestExposedType.getName)) |
| .safeSet(KubernetesConfigOptions.CLUSTER_ID, shutDownRequest.clusterId) |
| .safeSet( |
| KubernetesConfigOptions.CONTAINER_IMAGE, |
| shutDownRequest.kubernetesDeployParam.flinkImage) |
| .safeSet( |
| KubernetesConfigOptions.KUBE_CONFIG_FILE, |
| getDefaultKubernetesConf(shutDownRequest.kubernetesDeployParam.kubeConf)) |
| kubeClient = FlinkKubeClientFactory.getInstance.fromConfiguration(flinkConfig, "client") |
| val kubeClientWrapper = new FlinkKubernetesClient(kubeClient) |
| |
| val stopAndCleanupState = shutDownRequest.clusterId != null && kubeClientWrapper |
| .getService(shutDownRequest.clusterId) |
| .isPresent |
| if (stopAndCleanupState) { |
| kubeClient.stopAndCleanupCluster(shutDownRequest.clusterId) |
| ShutDownResponse(shutDownRequest.clusterId) |
| } else { |
| null |
| } |
| } catch { |
| case e: Exception => |
| logError(s"shutdown flink session fail in ${shutDownRequest.executionMode} mode") |
| e.printStackTrace() |
| throw e |
| } finally { |
| Utils.close(kubeClient) |
| } |
| } |
| |
| override def doTriggerSavepoint( |
| triggerSavepointRequest: TriggerSavepointRequest, |
| flinkConfig: Configuration): SavepointResponse = { |
| flinkConfig.safeSet( |
| DeploymentOptions.TARGET, |
| FlinkExecutionMode.KUBERNETES_NATIVE_SESSION.getName) |
| super.doTriggerSavepoint(triggerSavepointRequest, flinkConfig) |
| } |
| } |
