You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/01/18 06:54:22 UTC
[GitHub] [flink] wangyang0918 commented on a change in pull request #17098: [FLINK-15648][flink-kubernetes] Support to configure limit for CPU & memory
wangyang0918 commented on a change in pull request #17098:
URL: https://github.com/apache/flink/pull/17098#discussion_r786443356
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java
##########
@@ -122,6 +122,26 @@ public double getJobManagerCPU() {
return flinkConfig.getDouble(KubernetesConfigOptions.JOB_MANAGER_CPU);
}
+ public double getJobManagerCPULimitFactor() {
+ double limitFactor =
Review comment:
nit: `limitFactor` could be final.
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java
##########
@@ -122,6 +122,26 @@ public double getJobManagerCPU() {
return flinkConfig.getDouble(KubernetesConfigOptions.JOB_MANAGER_CPU);
}
+ public double getJobManagerCPULimitFactor() {
+ double limitFactor =
+ flinkConfig.getDouble(KubernetesConfigOptions.JOB_MANAGER_CPU_LIMIT_FACTOR);
+ checkArgument(
+ limitFactor >= 1,
+ "%s should be greater or equal to 1.",
+ KubernetesConfigOptions.JOB_MANAGER_CPU_LIMIT_FACTOR.key());
+ return limitFactor;
+ }
+
+ public double getJobManagerMemoryLimitFactor() {
+ double limitFactor =
Review comment:
nit: limitFactor could be final.
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java
##########
@@ -122,6 +122,26 @@ public double getTaskManagerCPU() {
.doubleValue();
}
+ public double getTaskManagerCPULimitFactor() {
+ double limitFactor =
Review comment:
nit: limitFactor could be final.
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java
##########
@@ -310,26 +310,33 @@ public static CompletedCheckpointStore createCompletedCheckpointStore(
*
* @param resourceRequirements resource requirements in pod template
* @param mem Memory in mb.
+ * @param memoryLimitFactor limit factor for the memory, used to set the limit resources.
* @param cpu cpu.
+ * @param cpuLimitFactor limit factor for the cpu, used to set the limit resources.
* @param externalResources external resources
* @param externalResourceConfigKeys config keys of external resources
* @return KubernetesResource requirements.
*/
public static ResourceRequirements getResourceRequirements(
ResourceRequirements resourceRequirements,
int mem,
+ double memoryLimitFactor,
double cpu,
+ double cpuLimitFactor,
Map<String, ExternalResource> externalResources,
Map<String, String> externalResourceConfigKeys) {
final Quantity cpuQuantity = new Quantity(String.valueOf(cpu));
+ final Quantity cpuLimitQuantity = new Quantity(String.valueOf(cpu * cpuLimitFactor));
final Quantity memQuantity = new Quantity(mem + Constants.RESOURCE_UNIT_MB);
+ final Quantity memQuantityLimit =
+ new Quantity(((int) (mem * memoryLimitFactor)) + Constants.RESOURCE_UNIT_MB);
Review comment:
Maybe we do not need to use the `int` here since K8s could also support decimal amount for memory resource. e.g. `1024.5m` `1.5g`. But the current implementation also make sense to me.
##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java
##########
@@ -122,6 +122,26 @@ public double getTaskManagerCPU() {
.doubleValue();
}
+ public double getTaskManagerCPULimitFactor() {
+ double limitFactor =
+ flinkConfig.getDouble(KubernetesConfigOptions.TASK_MANAGER_CPU_LIMIT_FACTOR);
+ checkArgument(
+ limitFactor >= 1,
+ "%s should be greater or equal to 1.",
+ KubernetesConfigOptions.TASK_MANAGER_CPU_LIMIT_FACTOR.key());
+ return limitFactor;
+ }
+
+ public double getTaskManagerMemoryLimitFactor() {
+ double limitFactor =
Review comment:
nit: limitFactor could be final.
##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
##########
@@ -116,6 +117,8 @@ protected void setupFlinkConfig() {
flinkConfig.set(JobManagerOptions.PORT, RPC_PORT);
flinkConfig.set(BlobServerOptions.PORT, Integer.toString(BLOB_SERVER_PORT));
flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_CPU, JOB_MANAGER_CPU);
+ flinkConfig.set(
Review comment:
I think we could leave the `KubernetesConfigOptions.JOB_MANAGER_CPU` as default since we do not verify the resource limit in the tests of `Fabric8FlinkKubeClientTest`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org