[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients #9965

wangyang0918 · 2019-10-22T05:22:45Z

What is the purpose of the change

This PR is part of FLINK-9953(Active Kubernetes integration). This PR introduces the FlinkKubeClient API and adds fabric8 implementation. The following k8s resource decorators are added to support for basic k8s integration.

Service, internal service is used for tm->jm hearbeat, rest service is used for flink client to submit job.
ConfigMap, save the flink-conf.yaml and log4j config files. It will be mounted into jm/tm pod.
Deployment, for job manager. When it crashes exceptionally, a new one will be started.
Pod, for task manager. It will created by KubernetesResourceManager.
OwnerReference, all resources are set owner reference to internal service. When the internal service is deleted, they will be deleted by gc.

This PR is based on #9695.

Brief change log

Add config options and cli options for kubernetes
Add FlinkKubeClient API and basic kubernetes resources
Add fabric8 kubeClient implementation and decorators
Add tests for fabric8 FlinkKubeClient implementation

Verifying this change

This change added Fabric8ClientTest public api implementation of FlinkKubeClient.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable)

flinkbot · 2019-10-22T05:26:01Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 685652d (Wed Dec 04 16:43:19 UTC 2019)

Warnings:

1 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2019-10-22T05:47:24Z

CI report:

6f90b45 : SUCCESS Build
86aa5ce : SUCCESS Build
7851d84 : FAILURE Build
d49bb0a : SUCCESS Build
0e88e0e : SUCCESS Build
9afcc7b : FAILURE Build
06e6b2b : SUCCESS Build
816a11a : SUCCESS Build
e806910 : SUCCESS Build
e85e76c : SUCCESS Build
d1c41ae : FAILURE Build
1e54eb3 : CANCELED Build
685652d : FAILURE Build
1b37e8e : CANCELED Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

...-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/fabric8/FlinkConfigMap.java

flink-kubernetes/pom.xml

...ernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java

...-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java

...n/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkMasterDeploymentDecorator.java

...src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/TaskManagerPodDecorator.java

wangyang0918 · 2019-10-23T07:06:01Z

I have addressed @MalcolmSanders 's comments and updated the PR.

KarmaGYZ

Thanks for open this PR, @wangyang0918 !
Good to see the K8S support in flink. Some comment left.

BTW, I think we could find a better way to deliver relevant configuration. Now, too many classes hold the flinkConfig.

...ernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java

...kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/TaskManagerPodParameter.java

...kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/ActionWatcher.java

...bernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/FlinkDeployment.java

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/Resource.java

...-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java

tisonkun

Thanks for your working @wangyang0918 ! I left some comments.

tisonkun · 2019-11-28T06:38:46Z

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java

+	/**
+	 * Create kubernetes config map, include flink-conf.yaml, log4j.properties.
+	 */
+	void createConfigMap() throws Exception;


I notice that some of "create" functions are blocking while others are non-blocking. What is the different between them and how is the decision made?

All the create operation is asynchronous, including createConfigMap createInternalService createFlinkMasterDeployment createTaskManagerPod. We just add a CompletableFuture for creating service because we need to make sure the service are created successfully. Then we could get the uuid of internal service and use the rest service to submit job.

tisonkun · 2019-11-28T06:42:50Z

...kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/ActionWatcher.java

+ */
+public class ActionWatcher<T extends HasMetadata> implements Watcher<T> {
+	private final CountDownLatch latch = new CountDownLatch(1);
+	private final AtomicReference<T> reference = new AtomicReference();


I don't see any atomic actions like compare-and-set or get-and-set are in used. Why do you instance an AtomicReference here? Is volatile sufficient?

I use AtomicReference just to make sure the atomic operation of set and get from multiple threads. Also volatile is enough here for object reference. AtomicReference is more clear, if you insist, i will use volatile instead.

tisonkun · 2019-11-28T06:43:22Z

...kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/ActionWatcher.java

+	private final CountDownLatch latch = new CountDownLatch(1);
+	private final AtomicReference<T> reference = new AtomicReference();
+	private final T resource;
+	private Action expectedAction;


It could be final IMO.

Will fix it.

tisonkun · 2019-11-28T06:46:31Z

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/Resource.java

+ */
+public abstract class Resource<T> {
+
+	protected T internalResource;


given we have getter for internalResource and flinkConfig they could be private final; just notice their setters are not into used at all.

There is code

subclass extends Resource { super(flinkConfig); this.internalResource = internalResource; }

which could be

super(flinkConfig, internalResource);

Nice suggestion. I will update the Resource and implementation class.

The setInternalResource will be kept. It will be used by Decorator. Extract real resource from resource, decorate and put it back.

tisonkun · 2019-11-28T06:50:11Z

...-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java

+
+	private final KubernetesClient internalClient;
+
+	private List<Decorator<ConfigMap, FlinkConfigMap>> configMapDecorators;


I'd tend to mark these lists as final when possible

Agree. Final will be better.

tisonkun · 2019-11-28T06:51:53Z

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java

+	/**
+	 * Get the rest endpoints for access outside cluster.
+	 */
+	Endpoint getRestEndpoints(String clusterId);


I notice that it could be Nullable in Faric8 implementation. The fact that it is nullable and the meaning of null should be documented.

I will add return description for getRestEndpoints and other methods may return null.

maybe getRestEndpoint since only one Endpoint would be returned

Make sense. I will fix it.

tisonkun · 2019-11-28T06:52:44Z

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java

+	/**
+	 * Get the kubernetes internal service of the given flink clusterId.
+	 */
+	FlinkService getInternalService(String clusterId);


I notice that it could be Nullable in Faric8 implementation. The fact that it is nullable and the meaning of null should be documented.

Will fix it.

tisonkun · 2019-11-28T06:54:42Z

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/KubeClientFactory.java

+			try {
+				config = Config.fromKubeconfig(KubernetesUtils.getContentFromFile(kubeConfigFile));
+			} catch (IOException e) {
+				LOG.error("Load kubernetes config failed.", e);


We don't throw exception here so new DefaultKubernetesClient(config); could be new DefaultKubernetesClient(null);. What is the semantic there then? It looks weird we log error but control flow continues.

An exception should be thrown if we get the content from kube config file error. I will fix it.

tisonkun · 2019-11-28T06:57:54Z

also you could rebase the branch on current master especially on merged FLINK-10932

wangyang0918 · 2019-11-29T12:51:30Z

@tisonkun Really thanks for your detailed review. I have rebase the master and addressed all your comments. Please take a look.

zhuzhurk

Thanks for open this PR @wangyang0918 .
I have a few comments for the configs and will later go through the remaining parts.

...ernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/Resource.java

...-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/FlinkService.java

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/FlinkPod.java

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java

...kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/LabelBuilder.java

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/Decorator.java

...etes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ConfigMapDecorator.java

...n/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkMasterDeploymentDecorator.java

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/KubeClientFactory.java

...-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java

zhuzhurk · 2019-12-02T03:49:16Z

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java

+	/**
+	 * Get the rest endpoints for access outside cluster.
+	 */
+	Endpoint getRestEndpoints(String clusterId);


maybe getRestEndpoint since only one Endpoint would be returned

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Endpoint.java

...-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java

wangyang0918 · 2019-12-02T06:34:21Z

@zhuzhurk Thanks a lot for your detailed review. I will address your comments one by one.

wangyang0918 · 2019-12-03T03:00:03Z

@tisonkun All the comments from you and @zhuzhurk have been addressed. Could you please take a look again and help to merge.

zhuzhurk

Thanks for addressing all the comments @wangyang0918 .
The change looks good to me except for some more minor comments.

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesUtilsTest.java

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java

wangyang0918 · 2019-12-03T08:00:17Z

@zhuzhurk I have addressed all your comments. Please take a look.

zhuzhurk · 2019-12-03T08:21:19Z

LGTM. +1 to merge it.

wangyang0918 · 2019-12-03T08:24:10Z

@tisonkun Please have a final check and help to merge. Really thanks for your kind help. -:)

tisonkun · 2019-12-04T13:47:50Z

...ernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java

+			"to the number of slots per TaskManager");
+
+	public static final ConfigOption<String> CONTAINER_IMAGE_PULL_POLICY =
+		key("kubernetes.container.image.pullPolicy")


for consistency, "pull-policy"?

I will fix it.

…rnetes. All the basic and necessary config options has been added. Such as, image name, service account, taskmanager resources, etc.

…s resources. The basic resources for active Kubernetes integration include ConfigMap, Deployment, Service, Pod.

…ecorators. The Fabric8FlinkKubeClient will be used to interact with Kubernetes API server to create/delete resources and get the status. The decorators will be used to build the complete Kubernetes resource description.

tisonkun

Thanks for your contribution @wangyang0918 . Merging now...I'll fix two typo as listed below. Be care to modify your following PRs.

tisonkun · 2019-12-05T02:34:54Z

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesCliOptions.java

+		.desc(KubernetesConfigOptions.CONTAINER_IMAGE.description().toString())
+		.build();
+
+	public static final Option JOB_MANAGER_MEMEORY_OPTION = Option.builder("jm")


Suggested change

public static final Option JOB_MANAGER_MEMEORY_OPTION = Option.builder("jm")

public static final Option JOB_MANAGER_MEMORY_OPTION = Option.builder("jm")

typo. I will fix on merging.

tisonkun · 2019-12-05T02:35:10Z

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesCliOptions.java

+		.desc("Memory for JobManager Container with optional unit (default: MB)")
+		.build();
+
+	public static final Option TASK_MANAGER_MEMEORY_OPTION = Option.builder("tm")


Suggested change

public static final Option TASK_MANAGER_MEMEORY_OPTION = Option.builder("tm")

public static final Option TASK_MANAGER_MEMORY_OPTION = Option.builder("tm")

typo. I will fix on merging.

tillrohrmann · 2020-01-31T13:23:49Z

...kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/ActionWatcher.java

+	}
+
+	@Override
+	public void onClose(KubernetesClientException e) {


Why is it ok to simply ignore the onClose event? Can't it happen that there is a problem (e.g. e != null)? Shouldn't we then signal this to potential waiters who have called await @wangyang0918?

rmetzger added the review=description? label Oct 22, 2019

wangyang0918 changed the title ~~Flink 10935~~ [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients Oct 22, 2019

rmetzger added the component=Deployment/Kubernetes label Oct 22, 2019

wangyang0918 force-pushed the FLINK-10935 branch from 6f90b45 to 86aa5ce Compare October 22, 2019 08:46

MalcolmSanders reviewed Oct 22, 2019

View reviewed changes

wangyang0918 force-pushed the FLINK-10935 branch from 86aa5ce to 7851d84 Compare October 23, 2019 07:02

wangyang0918 mentioned this pull request Oct 23, 2019

[FLINK-9955] Add Kubernetes ClusterDescriptor to support deploying session cluster. #9973

Closed

wangyang0918 force-pushed the FLINK-10935 branch from 7851d84 to d49bb0a Compare October 24, 2019 11:30

This was referenced Oct 24, 2019

[FLINK-10968][kubernetes] Implement TaskManager Entrypoint for Kubernetes. #9985

Closed

[FLINK-9495][kubernetes] Implement ResourceManager for Kubernetes. #9984

Merged

[FLINK-10933][kubernetes] Implement KubernetesSessionClusterEntrypoint. #9986

Merged

KarmaGYZ reviewed Oct 29, 2019

View reviewed changes

wangyang0918 force-pushed the FLINK-10935 branch 2 times, most recently from d88f5bb to 0e88e0e Compare October 29, 2019 12:47

This was referenced Nov 19, 2019

[FLINK-10937][dist] Add kubernetes-entry.sh and kubernetes-session.sh. #10246

Closed

[FLINK-10936][kubernetes] Implement Kubernetes command line tools to support session cluster. #10245

Closed

wangyang0918 force-pushed the FLINK-10935 branch 2 times, most recently from 9afcc7b to b3d9d02 Compare November 25, 2019 08:59

tisonkun reviewed Nov 28, 2019

View reviewed changes

wangyang0918 force-pushed the FLINK-10935 branch from 06e6b2b to 816a11a Compare November 29, 2019 12:47

wangyang0918 force-pushed the FLINK-10935 branch from 816a11a to e806910 Compare November 30, 2019 03:50

zhuzhurk reviewed Dec 1, 2019

View reviewed changes

zhuzhurk reviewed Dec 2, 2019

View reviewed changes

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Endpoint.java Show resolved Hide resolved

...-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java Outdated Show resolved Hide resolved

wangyang0918 force-pushed the FLINK-10935 branch from e806910 to e85e76c Compare December 2, 2019 11:41

zhuzhurk reviewed Dec 3, 2019

View reviewed changes

wangyang0918 force-pushed the FLINK-10935 branch from e85e76c to d1c41ae Compare December 3, 2019 07:59

wangyang0918 force-pushed the FLINK-10935 branch from d1c41ae to 1e54eb3 Compare December 4, 2019 08:59

tisonkun self-requested a review December 4, 2019 13:18

tisonkun reviewed Dec 4, 2019

View reviewed changes

wangyang0918 force-pushed the FLINK-10935 branch from 1e54eb3 to 685652d Compare December 4, 2019 14:47

wangyang0918 added 3 commits December 5, 2019 10:38

[FLINK-10935][kubernetes] Add config options and cli options for kube…

9ef99a7

…rnetes. All the basic and necessary config options has been added. Such as, image name, service account, taskmanager resources, etc.

[FLINK-10935][kubernetes] Add FlinkKubeClient API and basic kubernete…

16f8612

…s resources. The basic resources for active Kubernetes integration include ConfigMap, Deployment, Service, Pod.

wangyang0918 force-pushed the FLINK-10935 branch from 685652d to 1b37e8e Compare December 5, 2019 02:39

tisonkun approved these changes Dec 5, 2019

View reviewed changes

tisonkun closed this in f49e632 Dec 5, 2019

wangyang0918 deleted the FLINK-10935 branch December 5, 2019 05:52

tillrohrmann reviewed Jan 31, 2020

View reviewed changes


		private final KubernetesClient internalClient;

		private List<Decorator<ConfigMap, FlinkConfigMap>> configMapDecorators;

	public static final Option JOB_MANAGER_MEMEORY_OPTION = Option.builder("jm")
	public static final Option JOB_MANAGER_MEMORY_OPTION = Option.builder("jm")

	public static final Option TASK_MANAGER_MEMEORY_OPTION = Option.builder("tm")
	public static final Option TASK_MANAGER_MEMORY_OPTION = Option.builder("tm")

[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients #9965

[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients #9965

Conversation

wangyang0918 commented Oct 22, 2019 • edited

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Oct 22, 2019 • edited

Automated Checks

Review Progress

flinkbot commented Oct 22, 2019 • edited

CI report:

wangyang0918 commented Oct 23, 2019

KarmaGYZ left a comment

Choose a reason for hiding this comment

tisonkun left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wangyang0918 Nov 28, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tisonkun commented Nov 28, 2019

wangyang0918 commented Nov 29, 2019

zhuzhurk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wangyang0918 commented Dec 2, 2019

wangyang0918 commented Dec 3, 2019

zhuzhurk left a comment

Choose a reason for hiding this comment

wangyang0918 commented Dec 3, 2019

zhuzhurk commented Dec 3, 2019

wangyang0918 commented Dec 3, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tisonkun left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wangyang0918 commented Oct 22, 2019 •

edited

flinkbot commented Oct 22, 2019 •

edited

flinkbot commented Oct 22, 2019 •

edited

wangyang0918 Nov 28, 2019 •

edited