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.

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 List<URI>	checkJarFileForApplicationMode(Configuration configuration)
static CompletedCheckpointStore	createCompletedCheckpointStore(Configuration configuration, FlinkKubeClient kubeClient, Executor executor, String configMapName, String lockIdentity, int maxNumberOfCheckpointsToRetain, SharedStateRegistryFactory sharedStateRegistryFactory, Executor ioExecutor, RestoreMode restoreMode) Create a DefaultCompletedCheckpointStore with KubernetesStateHandleStore.
static void	createConfigMapIfItDoesNotExist(FlinkKubeClient flinkKubeClient, String configMapName, String clusterId) Creates a config map with the given name if it does not exist.
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 String	createSingleLeaderKey(String componentId)
static String	encodeLeaderInformation(LeaderInformation leaderInformation)
static String	extractLeaderName(String key)
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 String	getNamespacedServiceName(io.fabric8.kubernetes.api.model.Service service) Generate namespaced name of the service.
static KubernetesConfigMap	getOnlyConfigMap(List<KubernetesConfigMap> configMaps, String expectedConfigMapName) Check the ConfigMap list should only contain the expected one.
static io.fabric8.kubernetes.api.model.ResourceRequirements	getResourceRequirements(io.fabric8.kubernetes.api.model.ResourceRequirements resourceRequirements, int mem, double memoryLimitFactor, double cpu, double cpuLimitFactor, 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 boolean	isHostNetwork(Configuration configuration) Checks if hostNetwork is enabled.
static boolean	isSingleLeaderKey(String key)
static FlinkPod	loadPodFromTemplateFile(FlinkKubeClient kubeClient, File podTemplateFile, String mainContainerName)
static Optional<LeaderInformation>	parseLeaderInformationSafely(String value)
static Integer	parsePort(Configuration flinkConfig, ConfigOption<String> port) Parse a valid port for the config option.
static String	resolveDNSPolicy(String dnsPolicy, boolean hostNetworkEnabled) Resolve the DNS policy defined value.
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.

getOnlyConfigMap

public static KubernetesConfigMap getOnlyConfigMap(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,
                                                                      @Nullable
                                                                      String lockIdentity,
                                                                      int maxNumberOfCheckpointsToRetain,
                                                                      SharedStateRegistryFactory sharedStateRegistryFactory,
                                                                      Executor ioExecutor,
                                                                      RestoreMode restoreMode)
                                                               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

restoreMode - the mode in which the job is restoring

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 memoryLimitFactor,
                                                                                           double cpu,
                                                                                           double cpuLimitFactor,
                                                                                           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.

memoryLimitFactor - limit factor for the memory, used to set the limit resources.

cpu - cpu.

cpuLimitFactor - limit factor for the cpu, used to set the limit resources.

externalResources - external resources

externalResourceConfigKeys - config keys of external resources

Returns:

KubernetesResource requirements.

getStartCommandWithBashWrapper

public static List<String> getStartCommandWithBashWrapper(String command)

checkJarFileForApplicationMode

public static List<URI> 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

resolveDNSPolicy

public static String resolveDNSPolicy(String dnsPolicy,
                                      boolean hostNetworkEnabled)

Resolve the DNS policy defined value. Return DNS_POLICY_HOSTNETWORK if host network enabled. If not, check whether there is a DNS policy overridden in pod template.

Parameters:

dnsPolicy - DNS policy defined in pod template spec

hostNetworkEnabled - Host network enabled or not

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.

isHostNetwork

public static boolean isHostNetwork(Configuration configuration)

Checks if hostNetwork is enabled.

createConfigMapIfItDoesNotExist

public static void createConfigMapIfItDoesNotExist(FlinkKubeClient flinkKubeClient,
                                                   String configMapName,
                                                   String clusterId)
                                            throws FlinkException

Creates a config map with the given name if it does not exist.

Parameters:

flinkKubeClient - to use for creating the config map

configMapName - name of the config map

clusterId - clusterId to which the map belongs

Throws:

FlinkException - if the config map could not be created

encodeLeaderInformation

public static String encodeLeaderInformation(LeaderInformation leaderInformation)

parseLeaderInformationSafely

public static Optional<LeaderInformation> parseLeaderInformationSafely(String value)

createSingleLeaderKey

public static String createSingleLeaderKey(String componentId)

isSingleLeaderKey

public static boolean isSingleLeaderKey(String key)

extractLeaderName

public static String extractLeaderName(String key)

getNamespacedServiceName

public static String getNamespacedServiceName(io.fabric8.kubernetes.api.model.Service service)

Generate namespaced name of the service.