org.apache.flink.kubernetes.utils

Class KubernetesUtils

java.lang.Object

org.apache.flink.kubernetes.utils.KubernetesUtils

public class KubernetesUtils
extends Object

Common utils for Kubernetes.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KubernetesUtils.ClusterComponent Cluster components.

Method Summary

Modifier and Type	Method and Description
static void	checkAndUpdatePortConfigOption(Configuration flinkConfig, ConfigOption<String> port, int fallbackPort) Check whether the port config option is a fixed port.
static KubernetesConfigMap	checkConfigMaps(List<KubernetesConfigMap> configMaps, String expectedConfigMapName) Check the ConfigMap list should only contain the expected one.
static List<File>	checkJarFileForApplicationMode(Configuration configuration)
static CompletedCheckpointStore	createCompletedCheckpointStore(Configuration configuration, FlinkKubeClient kubeClient, Executor executor, String configMapName, String lockIdentity, int maxNumberOfCheckpointsToRetain) Create a DefaultCompletedCheckpointStore with KubernetesStateHandleStore.
static KubernetesStateHandleStore<JobGraph>	createJobGraphStateHandleStore(Configuration configuration, FlinkKubeClient flinkKubeClient, String configMapName, String lockIdentity) Create a KubernetesStateHandleStore which storing JobGraph.
static JobGraphStore	createJobGraphStore(Configuration configuration, FlinkKubeClient flinkKubeClient, String configMapName, String lockIdentity) Create a DefaultJobGraphStore with NoOpJobGraphStoreWatcher.
static Map<String,String>	getCommonLabels(String clusterId) Get the common labels for Flink native clusters.
static Map<String,String>	getConfigMapLabels(String clusterId, String type) Get ConfigMap labels for the current Flink cluster.
static String	getDeploymentName(String clusterId) Generate name of the Deployment.
static Map<String,String>	getJobManagerSelectors(String clusterId) Get job manager selectors for the current Flink cluster.
static LeaderInformation	getLeaderInformationFromConfigMap(KubernetesConfigMap configMap) Get the LeaderInformation from ConfigMap.
static io.fabric8.kubernetes.api.model.ResourceRequirements	getResourceRequirements(io.fabric8.kubernetes.api.model.ResourceRequirements resourceRequirements, int mem, double cpu, Map<String,ExternalResource> externalResources, Map<String,String> externalResourceConfigKeys) Get resource requirements from memory and cpu.
static String	getServiceAccount(FlinkPod flinkPod) Get the service account from the input pod first, if not specified, the service account name will be used.
static List<String>	getStartCommandWithBashWrapper(String command)
static File	getTaskManagerPodTemplateFileInPod()
static Map<String,String>	getTaskManagerSelectors(String clusterId) Get task manager selectors for the current Flink cluster.
static FlinkPod	loadPodFromTemplateFile(FlinkKubeClient kubeClient, File podTemplateFile, String mainContainerName)
static Integer	parsePort(Configuration flinkConfig, ConfigOption<String> port) Parse a valid port for the config option.
static <T> String	resolveUserDefinedValue(Configuration flinkConfig, ConfigOption<T> configOption, String valueOfConfigOptionOrDefault, String valueOfPodTemplate, String fieldDescription) Resolve the user defined value with the precedence.
static String	tryToGetPrettyPrintYaml(io.fabric8.kubernetes.api.model.KubernetesResource kubernetesResource) Try to get the pretty print yaml for Kubernetes resource.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

checkAndUpdatePortConfigOption

public static void checkAndUpdatePortConfigOption(Configuration flinkConfig,
                                                  ConfigOption<String> port,
                                                  int fallbackPort)

Check whether the port config option is a fixed port. If not, the fallback port will be set to configuration.

Parameters:

flinkConfig - flink configuration

port - config option need to be checked

fallbackPort - the fallback port that will be set to the configuration

parsePort

public static Integer parsePort(Configuration flinkConfig,
                                ConfigOption<String> port)

Parse a valid port for the config option. A fixed port is expected, and do not support a range of ports.

Parameters:

flinkConfig - flink config

port - port config option

Returns:

valid port

getDeploymentName

public static String getDeploymentName(String clusterId)

Generate name of the Deployment.

getTaskManagerSelectors

public static Map<String,String> getTaskManagerSelectors(String clusterId)

Get task manager selectors for the current Flink cluster. They could be used to watch the pods status.

Returns:

Task manager labels.

getJobManagerSelectors

public static Map<String,String> getJobManagerSelectors(String clusterId)

Get job manager selectors for the current Flink cluster.

Returns:

JobManager selectors.

getCommonLabels

public static Map<String,String> getCommonLabels(String clusterId)

Get the common labels for Flink native clusters. All the Kubernetes resources will be set with these labels.

Parameters:

clusterId - cluster id

Returns:

Return common labels map

getConfigMapLabels

public static Map<String,String> getConfigMapLabels(String clusterId,
                                                    String type)

Get ConfigMap labels for the current Flink cluster. They could be used to filter and clean-up the resources.

Parameters:

clusterId - cluster id

type - the config map use case. It could only be Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY now.

Returns:

Return ConfigMap labels.

checkConfigMaps

public static KubernetesConfigMap checkConfigMaps(List<KubernetesConfigMap> configMaps,
                                                  String expectedConfigMapName)

Check the ConfigMap list should only contain the expected one.

Parameters:

configMaps - ConfigMap list to check

expectedConfigMapName - expected ConfigMap Name

Returns:

Return the expected ConfigMap

getLeaderInformationFromConfigMap

public static LeaderInformation getLeaderInformationFromConfigMap(KubernetesConfigMap configMap)

Get the LeaderInformation from ConfigMap.

Parameters:

configMap - ConfigMap contains the leader information

Returns:

Parsed leader information. It could be LeaderInformation.empty() if there is no corresponding data in the ConfigMap.

createJobGraphStore

public static JobGraphStore createJobGraphStore(Configuration configuration,
                                                FlinkKubeClient flinkKubeClient,
                                                String configMapName,
                                                String lockIdentity)
                                         throws Exception

Create a DefaultJobGraphStore with NoOpJobGraphStoreWatcher.

Parameters:

configuration - configuration to build a RetrievableStateStorageHelper

flinkKubeClient - flink kubernetes client

configMapName - ConfigMap name

lockIdentity - lock identity to check the leadership

Returns:

a DefaultJobGraphStore with NoOpJobGraphStoreWatcher

Throws:

Exception - when create the storage helper

createJobGraphStateHandleStore

public static KubernetesStateHandleStore<JobGraph> createJobGraphStateHandleStore(Configuration configuration,
                                                                                  FlinkKubeClient flinkKubeClient,
                                                                                  String configMapName,
                                                                                  String lockIdentity)
                                                                           throws Exception

Create a KubernetesStateHandleStore which storing JobGraph.

Parameters:

configuration - configuration to build a RetrievableStateStorageHelper

flinkKubeClient - flink kubernetes client

configMapName - ConfigMap name

lockIdentity - lock identity to check the leadership

Returns:

a KubernetesStateHandleStore which storing JobGraph.

Throws:

Exception - when create the storage helper

createCompletedCheckpointStore

public static CompletedCheckpointStore createCompletedCheckpointStore(Configuration configuration,
                                                                      FlinkKubeClient kubeClient,
                                                                      Executor executor,
                                                                      String configMapName,
                                                                      String lockIdentity,
                                                                      int maxNumberOfCheckpointsToRetain)
                                                               throws Exception

Create a DefaultCompletedCheckpointStore with KubernetesStateHandleStore.

Parameters:

configuration - configuration to build a RetrievableStateStorageHelper

kubeClient - flink kubernetes client

configMapName - ConfigMap name

executor - executor to run blocking calls

lockIdentity - lock identity to check the leadership

maxNumberOfCheckpointsToRetain - max number of checkpoints to retain on state store handle

Returns:

a DefaultCompletedCheckpointStore with KubernetesStateHandleStore.

Throws:

Exception - when create the storage helper failed

getResourceRequirements

public static io.fabric8.kubernetes.api.model.ResourceRequirements getResourceRequirements(io.fabric8.kubernetes.api.model.ResourceRequirements resourceRequirements,
                                                                                           int mem,
                                                                                           double cpu,
                                                                                           Map<String,ExternalResource> externalResources,
                                                                                           Map<String,String> externalResourceConfigKeys)

Get resource requirements from memory and cpu.

Parameters:

resourceRequirements - resource requirements in pod template

mem - Memory in mb.

cpu - cpu.

externalResources - external resources

externalResourceConfigKeys - config keys of external resources

Returns:

KubernetesResource requirements.

getStartCommandWithBashWrapper

public static List<String> getStartCommandWithBashWrapper(String command)

checkJarFileForApplicationMode

public static List<File> checkJarFileForApplicationMode(Configuration configuration)

loadPodFromTemplateFile

public static FlinkPod loadPodFromTemplateFile(FlinkKubeClient kubeClient,
                                               File podTemplateFile,
                                               String mainContainerName)

getTaskManagerPodTemplateFileInPod

public static File getTaskManagerPodTemplateFileInPod()

resolveUserDefinedValue

public static <T> String resolveUserDefinedValue(Configuration flinkConfig,
                                                 ConfigOption<T> configOption,
                                                 String valueOfConfigOptionOrDefault,
                                                 @Nullable
                                                 String valueOfPodTemplate,
                                                 String fieldDescription)

Resolve the user defined value with the precedence. First an explicit config option value is taken, then the value in pod template and at last the default value of a config option if nothing is specified.

Type Parameters:

T - The type of value associated with the configuration option.

Parameters:

flinkConfig - flink configuration

configOption - the config option to define the Kubernetes fields

valueOfConfigOptionOrDefault - the value defined by explicit config option or default

valueOfPodTemplate - the value defined in the pod template

fieldDescription - Kubernetes fields description

Returns:

the resolved value

getServiceAccount

@Nullable
public static String getServiceAccount(FlinkPod flinkPod)

Get the service account from the input pod first, if not specified, the service account name will be used.

Parameters:

flinkPod - the Flink pod to parse the service account

Returns:

the parsed service account

tryToGetPrettyPrintYaml

public static String tryToGetPrettyPrintYaml(io.fabric8.kubernetes.api.model.KubernetesResource kubernetesResource)

Try to get the pretty print yaml for Kubernetes resource.

Parameters:

kubernetesResource - kubernetes resource

Returns:

the pretty print yaml, or the KubernetesResource#toString() if parse failed.