1. flink k8s源码解析
   1. 客户端提交命令

./bin/kubernetes-session.sh \

  -Dkubernetes.namespace=flink-session-cluster \

  -Dkubernetes.jobmanager.service-account=flink \

  -Dkubernetes.cluster-id=session001 \

  -Dtaskmanager.memory.process.size=8192m \

  -Dkubernetes.taskmanager.cpu=1 \

  -Dtaskmanager.numberOfTaskSlots=4 \

  -Dresourcemanager.taskmanager-timeout=3600000

分析：

KubernetesSessionCli类就是上述命令行的入口类，首先从FlinkKubeClientFactory工厂类中选择Fabric8FlinkKubeClient实例类，在Fabric8FlinkKubeClient实例类中创建JM组件，TM pod，并且停止某个POD，停止cluster，创建configmap等。

 

1. 1. 创建JM组件
      1. 代码分析

调用关系：

一. 客户端输入命令后，先将命令解析成Confiugration。入口类为KubernetesSessionCli.java

其中getClusterSpecification代码如下所示：

根据客户端命令行的入参可以获取cluster的内存/slot个数配置，组装成一个ClusterSpecification对象。

二. 上面得到的ClusterSpecification对象作为deploySessionCluster的入参，继续进行内部cluster的部署（调用deployClusterInternal）。

三. deployClusterInternal的具体实现：
3.1 先填充flinkConf参数。填充的参数包括internal.cluster.execution-mode=DETACHED/NORMAL、kubernetes.internal.jobmanager.entrypoint.class=$entryPoint、blob.server.port=6124、taskmanager.rpc.port=6122、rest.bind-port=8081。如果启用了高可用模式，则设置high-availability.cluster-id=$cluster_id。

其中，核心代码如下所示：

3.2 具体逻辑如下所示：

1. 先获取jobmanger模板文件：是由kubernetes.pod-template-file.jobmanager指定，如果不存在。

2. 根据模板文件，创建FlinkPod对象。如果模板文件不存在，则创建一个空的FlinkPod对象。

3. 检查模板中是否设置了主container，如果设置了，则设置主container和other container。

4. buildKubernetesJobManagerSpecification的入参podTemplate和jobmanager配置（包括上游传递过来的JM 内存、cpu、slot个数配置以及flink conf中配置），返回一个KubernetesJobManagerSpecification实例。其中KubernetesJobManagerSpecification的生成规则，生成配置来源文件一个是pod模板文件里面生成和JM自定义参数组成，后者可以填充前者生成的FlinkPod（将mount、环境变量、cmdjob、hadoop配置文件、kerberos配置文件等都填充到模板生成的FlinkPod里面）。
KubernetesJobManagerSpecification的结构体如下，包含了Deployment以及Configmap和Service对象。

 

5. buildKubernetesJobManagerSpecification：总共包含了若干个装饰器

5.1 mount装饰器，将flink conf中的kubernetes.flink.conf.dir进行mount。

5.2 kerberos装饰器，根据flink conf中的security.kerberos.krb5-conf.path进行mount

5.3 hadoop配置装饰器。将宿主机上的hadoop配置进行mount

5.4 外部service装饰器。构建外部service

5.5 启动命令装饰器。主要是jobmanager启动的命令。

5.6 证书装饰器。主要是证书文件mount。

5.7 JM初始化装饰器。

6. 根据配置生成Deployment对象

可以上图，可以看到Deployment的各个属性的来源
其中:

apiversion=apps/v1(固定值)；

name=$cluster_id

labels=命令行入参kubernetes.jobmanager.labels的值；

replicas=命令行入参kubernetes.jobmanager.replicas的值；

metadata和spec来源于模板文件配置以及入参生成的FlinkPod中metadata以及spec规格。

7.最后会直接创建JM组件。

client.createJobManagerComponent(kubernetesJobManagerSpec);

创建JM组件，逻辑如下：

createJobManagerComponent(KubernetesJobManagerSpecification kubernetesJMSpec)

KubernetesJobManagerSpecification的结构体如下，包含了Deployment以及Configmap和Service对象。

 

调用k8s client创建jobmanager。这个功能类似

kubectl create -f jobmanager_session_xx.yaml

setOwnerReference将deploy与configmap与service绑定。

调用k8s client创建configmap与service。

。

1. 1. 1. 疑难问题

1. 需要在k8s CR上添加cpu和内存限制，如何添加？

解析：在CR里面有一个container，container有resource属性，

如下举例说明：

container:

"resource":{"limit":["cpu":{"amount":1}, "memory":{"amount":1}],"limit":["cpu":{"amount":1}, "memory":{"amount":1}]}

 

2. 在k8s 原生CR上和native k8s上添加cpu、内存限制有什么不同？

如果CR提供的标准的模板，每个application的CR都一样，那么我们可以使用标准的CR。

但是对于有定制化要求的CR，比如对JM、TM cpu和内存可以用户自定义，那么我们可以使用用户的自定义参数装饰一下原生的CR，使其成为一个新的CR，然后提交给k8s。

而native k8s提供了一套装饰的解决方案，可以使用用户输入的参数对CR原生模板进行装饰，然后提交到k8s执行。

 

1. 1. 1. Deployment数据结构

JM创建的时候依赖Deployment数据结构，Deployment如下所示：

 

 

1. 1. 创建TM POD
      1. 代码分析

核心代码如下：

1. TM模板文件，首先检查/opt/flink/pod-template/taskmanager-pod-template.yaml（固定文件目录，不可修改）是否存在。如果存在的话，根据这个TM模板生成TM FlinkPod。

2. 如果上述模板文件不存在，则创建一个新的TM FlinkPod文件。

3. 根据TM的入参（在命令行添加的入参）对FlinkPod进行修饰，修饰种类包括InitTaskManager修饰器、环境变量修饰器、证书修饰器、命令行修饰器、kerberos认证修饰器、flinkConf修饰器。最终生成一个KubernetesPod对象。

4. 根据KubernetesPod对象，调用createTaskManagerPod创建TM pod。

createTaskManagerPod函数如下所示：

1. 获取指定deploymentName的deployment实例。

2. deployment和resource进行绑定

3. 创建pod

 

1. 1. 1. 核心数据结构

 

 

 

1. Flink operator源码解析
   1. 概述

Flink k8s部署模式主要包括两种，一种是standalone模式，一种是native模式:

1. standalone模式，以独立模式在k8s上部署集群；

2. native模式，使用Flink原生Kubernetes支持部署Flink。仅支持更新版本的Flink。

使用方法：在CR的spec的下一级添加mode：native/standalone

Native模式可以支持session和application；

目前，Flink 支持多种 K8S 上的部署模式：

Standalone 部署模式：与物理机上的 Standalone 部署模式类似，用户需要自行维护所有资源的创建和管理。

Operator 部署模式：根据用户的配置，基于 Flink Operator 自动管理 Flink 作业及其生命周期。

Native 部署模式：类似于Yarn上的部署模式，JobManager 根据作业的需求自主创建和管理 TaskManager Pod。

1. 1. 创建Session
      1. 客户端提交CRD

通过CRD的方式提交session和job；

如下所示：

可以看到Kind类型为FlinkDeployment，表示新的Session的生成。如果CR中包含job，则为application模式，如果不包含，则为session模式。

1. 1. 1. CR对应的类

总的类是FlinkDeployment，其中继承了AbstractFlinkResource(继承CustomResource)。

其中大的框架包括标签："apiVersion", "kind", "metadata", "spec", "status"；

其中spec标签对应的类是FlinkDeploymentSpec；

Status标签对应的类是FlinkDeploymentStatus；

FlinkDeploymentSpec类如下所示：

 

1. 1. 1. Standalone session源码分析
2. FlinkOperator依然是入口，所有的提交任务以及对象初始化都是从这个地方开始。这儿会将用户的输入CR解析成FlinkDeployment对象。
3. FlinkDeploymentController是FlinkDeployment的控制器，负责FlinkDeployment对象的处理。
4. 检查运行模式mode，规则如下：如果CR中包含了job，则是application模式，如果不包含作业，则是session模式。

   

5. 根据协调器工厂类（ReconcilerFactory）获取（Reconciler），如果mode是application，则Reconciler类型为ApplicationReconciler，否则为SessionReconciler。
6. SessionReconciler处理逻辑：reconcileSpecChange->deploy->submitSessionCluster-> deploySessionCluster（StandaloneFlinkService.java）-> submitClusterInternal（StandaloneFlinkService.java）-> deploySessionCluster（KubernetesStandaloneClusterDescriptor.java）;其中Configuration也是根据FlinkDeployment提取得来的。
   StandaloneFlinkService.java：
   standalone也分为application cluster和session cluster。

   

KubernetesStandaloneClusterDescriptor.java：

 

其中deploySessionCluster的入参是ClusterSpecification，这个入参是根据FlinkDeployment间接提取出来的由内存和slot个数组合的对象。

ClusterSpecification类型：

可以看到ClusterSpecification对象中只有内存配置和每个TM包含的slot个数。打分

1. 终于到核心代码了，开始内部部署session。

   

   
   flinkConf参数的设置如下：
   设置flink conf的参数，blob.server.port=6124
   设置flink conf的参数，taskmanager.rpc.port=6122
   设置flink conf的参数，rest.bind-port=8081
   如果开启了高可用性，则设置high-availability.cluster-id=$clusterId。（这个标志是为了区别不同的cluster集群，在yarn场景会自动推断的）。
   接下来是调用getJobManagerSpec获取JM的spec规格。

   

   
   首先检查flink conf中的参数kubernetes.pod-template-file.jobmanager指定的JM模板文件，如果指定了则按照模板文件构建FlinkPod，如果没有指定，则新建一个新的FlinkPod。

• 然后根据kubernetesJobManagerParameters装饰FlinkPod。kubernetesJobManagerParameters是由flinkconf配置（由FlinkDeployment内的flink配置参数以及默认填充配置组成）以及JM配置组成。

  

  
  在buildKubernetesJobManagerSpecification中我们又看到了装饰器模式。
  装饰器列表如下所示：
• mount装饰器，将flink conf中的kubernetes.flink.conf.dir进行mount。
• kerberos装饰器，根据flink conf中的security.kerberos.krb5-conf.path进行mount
• hadoop配置装饰器。将宿主机上的hadoop配置进行mount
• 外部service装饰器。构建外部service
• 启动命令装饰器。主要是jobmanager启动的命令。
• 证书装饰器。主要是证书文件mount。
• JM初始化装饰器。

 

1. 最后调用createJobManagerComponent。这个是flink源码里面的代码了。
2. 最后调用createTaskManagerDeployment。这个也是flink源码里面的代码了。

 

1. 1. 1. Native session源码分析
2. flinkOperator依然是入口，所有的提交任务以及对象初始化都是从这个地方开始。这儿会将用户的输入CR解析成FlinkDeployment对象。
3. FlinkDeploymentController是FlinkDeployment的控制器，负责FlinkDeployment对象的处理。
4. 检查运行模式mode，规则如下：如果CR中包含了job，则是application模式，如果不包含作业，则是session模式。

   

5. 根据协调器工厂类（ReconcilerFactory）获取（Reconciler），如果mode是application，则Reconciler类型为ApplicationReconciler，否则为SessionReconciler。
6. SessionReconciler处理逻辑：reconcileSpecChange->deploy->submitSessionCluster-> deploySessionCluster(NativeFlinkService.java)-> submitClusterInternal(NativeFlinkService.java);其中Configuration也是根据FlinkDeployment提取得来的。

   

   
   其中deploySessionCluster的入参是ClusterSpecification，这个入参是根据FlinkDeployment间接提取出来的由内存和slot个数组合的对象。

ClusterSpecification类型：

可以看到ClusterSpecification对象中只有内存配置和每个TM包含的slot个数。打分

 

   注意：native模式下，deploySessionCluster是flink组件里面的函数了，也就是说直接调用flink组件来完成session部署。

 

1. 1. Session的删除

代码SessionReconciler.java。

功能：

1. 首先检查session里面是否有作业，则在一段时间后重新检查是否可以删除Session。

2. 如果session里面没有作业，则直接删除cluster。

接着调用deleteClusterDeployment函数。

接下来删除cluster。

org/apache/flink/kubernetes/operator/service/NativeFlinkService.java

 

1. 1. Session升级

功能：

1. 如果是第一次部署，则直接部署session；

2. 如果直接升级session（即session规格发生了变化），则需要先kill掉session，然后重新部署session。

代码：

入口函数：

AbstractFlinkResourceReconciler.java reconcile(FlinkResourceContext<CR> ctx)

 

reconcileSpecChange函数：先删除session cluster，然后重新部署（deploy）;

 

1. 1. Session Job的提交
      1. 触发条件

下面是输入的CR，表示只会提交session job，不会创建session；

可以看到Kind类型为FlinkSessionJob，表示只会提交session job；

1. 1. 1. 源码分析

在SessionJobReconciler.java：deploy函数中，调用submitJobToSessionCluster将作业提交到cluster中。

1. 先通过rest api: /jars/upload 将jar包上传到cluster集群上；

2. 然后通过/jars/:jarid/run提交job到cluster集群上；

 

1. 1. Session job的取消

问题：

1. 何时触发作业的取消？

2. 何时触发作业的升级？

 

创建资源：kubectl create -f deployment.yaml
应用或者更新资源配置：kubectl apply -f deployment.yaml

删除资源：kubectl delete -f deployment.yaml

 

1. 1. 1. 触发条件

对于停止作业，包括两种模式，一个是无状态停止作业，一个是savepoint触发停止。具体源码如下：

 

触发条件：

kubectl delete -f sessionjob.yaml

1. 1. 1. 源码分析

src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java

src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java

src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java

对于stopWithSavepoint和cancel函数都是flink源码对外提供的api函数。源码路径：src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java。具体如下所示：

 

 

1. 1. Session Job升级
      1. 触发升级的条件：

1. jobSpec发生了变化，即CR中的job规格发生了变化。

 

1. 1. 1. 源码分析

主逻辑：

reconcile(FlinkResourceContext<CR> ctx) throws Exception

当感受到CR发生变化的时候，触发job/session的升级。

调用reconcileSpecChange：（src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java）

核心逻辑：

1. 如果作业是按照savepoint升级，则先cancel（携带savepoint），然后重新部署job。

2. 如果作业正在运行中，则cancel job，作业状态就变成了suspended；

3. 然后对suspended的作业进行restore job；

 

具体代码：

 

1. 1. Application模式

总结：直接调用的flink源码进行实现的。

源码分析：

NativeFlinkService.java：

直接调用flink源码里面的ApplicationClusterDeployer里面的run函数，具体函数如下：

 

1. Dinky k8s提交flink任务源码解析
   1. 概述

Dinky支持以下几种 Flink on k8s 运行模式：

- Native-Kubernetes Application

- Native-Kubernetes Session

- Kubernetes Operator Application(基于官方operator实现)

 

### Kubernetes Operator 执行模式

 

:::tip 特点

使用 Kubernetes Operator 对 Flink 集群进行管理,可以实现集群的动态伸缩。

**特点：自动化管理，Operator 可以自动化管理应用程序的生命周期和运维任务；高度可扩展，Operator 允许开发者扩展 Kubernetes

的功能，以适应特定应用程序或工作负载的需求；智能决策和自愈能力，Operator

可以通过监视和分析应用程序的状态，做出智能决策以确保应用程序的健康和高可用性；自定义资源定义（CRD），Operator

使用自定义资源定义（CRD）来扩展 Kubernetes API，以支持新的自定义资源类型。

 

适用于：作业资源隔离，启动慢，动态扩容，节约网络资源，资源占用较多的批作业和流作业。

 

1. 1. Native-Kubernetes Application

KubernetesApplicationGateway.java

1. 首先按照application模式部署（直接给flink组件发送请求）；

2. 状态检查。首先检查POD是否启动，如果pod没有就绪，则5秒后再次检查。

3. 生成clusterid。首先获取第一个作业的jobid，clusterid=配置的clusterid+jobid；

 

1. 1. Native-Kubernetes Session

总结：

native session的作用就是调用flink源码里面的deploySessionCluster创建一个JM。它的输入主要是JM的内存、TM的内存、TM的slot个数。

注意：operator的代码创建native session的时候直接调用的是deploySessionCluster，而本开源服务直接调用的是deploySessionCluster（内部也会直接调用createJobManagerComponent）。

源码分析如下所示：

KubernetesSessionGateway.java：deployCluster函数，调用deloySessionCluster实际上是调用flink源码里面的KubernetesClusterDescriptor.deploySessionCluster(ClusterSpecification clusterSpecification)。

分析上面函数的55行，可以看出来JM/TM的内存设置均为页面的输入（即JobManagerOptions/TaskManagerOptions）。

1. 1. Kubernetes Operator Application

注意：

1. operator application只是提交FlinkDeployment CR；

2. 由于本服务需要依赖开源的flink operator服务，所以需要根据开源operator搭建一个，要不然进行任务的提交。

源码：

KubernetsOperatorGateway.java

这个主要是构建FlinkDeployment CR；

KubetnetsApplicationOperatorGateway.java

1. 首先获取到k8s client；

2. 获取组装好的FlinkDeployment对象；

3. k8s提交任务，类似于kubectl create -f xxx.yaml