New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

SAMZA-2067: Support Samza's running on Kubernetes #1197

Open

weiqingy wants to merge 10 commits into apache:master from weiqingy:samzak8rb

Contributor

weiqingy commented Oct 19, 2019 •

edited

Loading

What changes were proposed in this pull request?

Make Samza jobs able to run on Kubernetes natively.

How was this patch tested?

Run Samza jobs on AKS.

weiqingy changed the title ~~Support Samza's running on Kubernetes~~ [WIP] Support Samza's running on Kubernetes

weiqingy changed the title ~~[WIP] Support Samza's running on Kubernetes~~ [WIP] SAMZA-2067: Support Samza's running on Kubernetes

weiqingy changed the title ~~[WIP] SAMZA-2067: Support Samza's running on Kubernetes~~ [WIP] SAMZA-2067: Support Samza to run on Kubernetes

weiqingy changed the title ~~[WIP] SAMZA-2067: Support Samza to run on Kubernetes~~ [WIP] SAMZA-2067: Support Samza's running on Kubernetes

weiqingy force-pushed the samzak8rb branch from 2036e37 to ec9a40c Compare

October 20, 2019 03:58

jian-he force-pushed the samzak8rb branch from ec9a40c to fa25467 Compare

October 20, 2019 05:31


          Support Samza on Kubernetes. Contributed by Weiqing Yang &Jian He

381a683

jian-he force-pushed the samzak8rb branch from fa25467 to 381a683 Compare

October 20, 2019 05:36


          Fix the conflicts with master branch

5c835bf

xinyuiscool self-requested a review

October 24, 2019 18:47

Weiqing Yang added 2 commits

November 4, 2019 17:59


          clean code and add code comments

9d8e752


          Remove unused comments

1bbc8e0

weiqingy changed the title ~~[WIP] SAMZA-2067: Support Samza's running on Kubernetes~~ SAMZA-2067: Support Samza's running on Kubernetes

xinyuiscool reviewed

View reviewed changes

Contributor

xinyuiscool left a comment

Finish about 1/3 of it. Need to sync up to understand the design more.

samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java Outdated

@@ @@ -413,7 +413,8 @@ public static void main(String[] args) { @@
                     //Read and parse the coordinator system config.
                     LOG.info("Parsing coordinator system config {}", coordinatorSystemEnv);
                     coordinatorSystemConfig =
-                        new MapConfig(SamzaObjectMapper.getObjectMapper().readValue(coordinatorSystemEnv, Config.class));
+                        new MapConfig(SamzaObjectMapper.getObjectMapper()
+                            .readValue(coordinatorSystemEnv.replace("\\\"", "\""), Config.class));

Contributor

xinyuiscool Nov 9, 2019

Is there any reason we do this replacement?

Contributor Author

weiqingy Mar 5, 2021

The RB has been updated and aligned with the master branch. The change here has gone.

samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java Outdated

@@ @@ -93,7 +93,7 @@ @@
                  * The Allocator matches requests to resources and executes processes.
                  */
                 private final ContainerAllocator containerAllocator;
-                private final Thread allocatorThread;
+                private Thread allocatorThread = null;

Contributor

xinyuiscool Nov 9, 2019

please keep this final and assign to different values in the constructor.

Contributor Author

weiqingy Mar 5, 2021

Sure. The RB was updated, and the change here has gone.

samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java

-                  this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
-                  LOG.info("Finished container process manager initialization.");
+                  if (shouldStartAllocateThread()) {
+                    this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");

Contributor

xinyuiscool Nov 9, 2019

Do actually use containerAllocator anymore? Please walk through this part of the code with me offline.

samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java

+                // In Kubernetes, the pod will be started by kubelet automatically once it is allocated, it does not need a
+                // separate thread to keep polling the allocated resources to start the container.
+                public boolean shouldStartAllocateThread() {
+                  return !clusterResourceManager.getClass().getSimpleName().equals("KubeClusterResourceManager");

Contributor

xinyuiscool Nov 9, 2019

This lookup of specific cluster manager seems pretty hard to maintain. Can we think of a better to distinguish when we need this to do container allocation?

samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java Outdated

+                  LOG.debug("ContainerProcessManager state: Completed containers: {}, Configured containers: {}, Are there too many failed containers: {}",
+                    state.completedProcessors.get(), state.processorCount, jobFailureCriteriaMet ? "yes" : "no");
+                  if (shouldStartAllocateThread()) {

Contributor

xinyuiscool Nov 9, 2019

what does this if do?

Contributor Author

weiqingy Mar 5, 2021

This If is just for logging the aliveness info "Is allocator thread alive". The RB has been updated. The aliveness info is now in the LOG.debug() in line 207 (keep the code same as the current codebase).

samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java Outdated

-                  return jobFailureCriteriaMet || state.completedProcessors.get() == state.processorCount.get() || !allocatorThread.isAlive();
+                  boolean shouldShutdown = jobFailureCriteriaMet || state.completedProcessors.get() == state.processorCount.get();
+                  return shouldStartAllocateThread() ? shouldShutdown || !allocatorThread.isAlive() : shouldShutdown;

Contributor

xinyuiscool Nov 9, 2019

I guess it's nicer to do a null check of the thread instead of doing shouldStartAllocateThread()

Contributor Author

weiqingy Mar 5, 2021

right. updated.

samza-core/src/main/scala/org/apache/samza/coordinator/JobModelManager.scala Outdated

@@ @@ -23,34 +23,22 @@ import java.util @@
               import java.util.concurrent.atomic.AtomicReference
               import org.apache.samza.{Partition, SamzaException}
-              import org.apache.samza.config._

Contributor

xinyuiscool Nov 9, 2019

Seems all the changes in this file is just to reorg imports. If that's the case, I think this file can be left untouched.

Contributor Author

weiqingy Mar 5, 2021

The RB has been updated and aligned with the master branch. The change here has gone.

xinyuiscool reviewed

View reviewed changes

Contributor

xinyuiscool left a comment •

edited

Loading

Three main things:

the integration with current ContainerAllocator is quite hard to understand. It's more like just try to fit in instead of organic part of it. I would suggest to think about whether the current ContainerAllocator can be split into two allocator, one sync and one async. The yarn can use the Async allocator, and kubernetes can use the sync one. The thread can be brought up for the async allocator.
error handling: there seems to be quite a few loose ends about failures, e.g. operatorPod watcher failure, create failure, delete failure. All these need to be tied up so it can provide production quality code.
unit tests are completely missing. I expect unit test for new classes and new methods as possible.

Minor:

please add more comments and javadocs too.

samza-kubernetes/src/main/java/org/apache/samza/config/KubeConfig.java Outdated

+                // the image name of samza
+                public static final String APP_IMAGE = "kube.app.image";
+                public static final String DEFAULT_IMAGE = "weiqingyang/samza:v0";

Contributor

xinyuiscool Nov 15, 2019

Probably not put your personal id as the default config.

Contributor Author

weiqingy Mar 5, 2021

right. Updated it to "samza/samza:v0".

Contributor Author

weiqingy Mar 5, 2021

Will create samza docker hub if needed.

samza-kubernetes/src/main/java/org/apache/samza/config/KubeConfig.java Outdated

+                public static final String DEFAULT_DIRECTORY = "/opt/samza/";
+                // the memory and the cpu cores of container
+                public static final String CLUSTER_MANAGER_CONTAINER_MEM_SIZE = "cluster-manager.container.memory.mb";

Contributor

xinyuiscool Nov 15, 2019

Seems these configs should already be defined in the cluster manager config. If not, can we make sure it's shared?

Contributor Author

weiqingy Mar 5, 2021

RB was updated to remove these duplicate config definitions.

samza-kubernetes/src/main/java/org/apache/samza/job/kubernetes/KubeJob.java Outdated

+               */
+              public class KubeJob implements StreamJob {
+                private static final Logger LOG = LoggerFactory.getLogger(KubeJob.class);
+                private Config config;

Contributor

xinyuiscool Nov 15, 2019

mark all these vars final.

Contributor Author

weiqingy Mar 5, 2021

Done in both KubeJob and KubeClusterResourceManager.

samza-kubernetes/src/main/java/org/apache/samza/job/kubernetes/KubeJob.java Outdated

+                  this.kubernetesClient = KubeClientFactory.create();
+                  this.config = config;
+                  this.podName = String.format(JC_POD_NAME_FORMAT, JC_CONTAINER_NAME_PREFIX,
+                          config.get(APP_NAME, "samza"), config.get(APP_ID, "1"));

Contributor

xinyuiscool Nov 15, 2019

Please use ApplicaitonConfig.getAppName()/getAppId().

Contributor Author

weiqingy Mar 5, 2021

done

samza-kubernetes/src/main/java/org/apache/samza/config/KubeConfig.java

+              /**
+               * Kubernetes related configurations
+               */
+              public class KubeConfig {

Contributor

xinyuiscool Nov 15, 2019

I would suggest following the other Config class example to add getters for common configs, like namespace, jcPodName, etc. It's easier to centralize the place where these things are created. The rest of the code should just use these getters normally instead of directly accessing the config vars. It will be even better to make those static vars private.

Contributor Author

weiqingy Mar 5, 2021

sounds good. will update this class.

samza-kubernetes/src/main/java/org/apache/samza/job/kubernetes/KubeClusterResourceManager.java

+                  return cmdBuilder.buildCommand();
+                }
+                private CommandBuilder getCommandBuilder(String containerId) {

Contributor

xinyuiscool Nov 15, 2019

this method should be shared with Yarn I believe.

samza-kubernetes/src/main/java/org/apache/samza/job/kubernetes/KubeClusterResourceManager.java

+                }
+                // Construct the envs for the task container pod
+                private List<EnvVar> getEnvs(CommandBuilder cmdBuilder) {

Contributor

xinyuiscool Nov 15, 2019

Can we make it an static util method?

samza-kubernetes/src/main/java/org/apache/samza/job/kubernetes/KubePodStatusWatcher.java

+               */
+              // TODO: SAMZA-2369: Add a logging thread which is similar to LoggingPodStatusWatcher in Spark
+              public class KubePodStatusWatcher implements Watcher<Pod> {

Contributor

xinyuiscool Nov 15, 2019

Seems we already have a watcher in job coordinator. Any idea what this thing does? I don't see it's been used.

samza-kubernetes/src/main/java/org/apache/samza/job/kubernetes/KubeUtils.java

+              /**
+               * Convenient utility class with static methods.
+               */
+              public class KubeUtils {

Contributor

xinyuiscool Nov 15, 2019

For each method, you need to put javadocs and add tests.

samza-kubernetes/src/main/java/org/apache/samza/job/kubernetes/KubeUtils.java


		package org.apache.samza.job.kubernetes;

		import io.fabric8.kubernetes.api.model.*;

Contributor

xinyuiscool Nov 15, 2019

Please make sure NOT use wild card imports for all the source files.

Contributor Author

weiqingy commented Nov 26, 2019 •

edited

Loading

@xinyuiscool Thanks for the reviewing. I'll update the PR to fix the comments soon.

Contributor

mynameborat commented Apr 10, 2020

@weiqingy Is this still being worked on?

Contributor Author

weiqingy commented Apr 10, 2020

Hi @mynameborat Yes, we are actively working on it, targeting Samza 1.5.

Weiqing Yang added 6 commits

February 22, 2021 13:17


          Merge remote-tracking branch 'upstream/master' into samzak8rb

aafc2d4


          Fix conflicts

9071f7b


          fix conflict

6ff8d7c


          Fix conflicts

66cb923


          Fix build error

fb405ae


          Fix review comments

44bbb47

rohitverma02 commented Jun 8, 2021

@weiqingy Samza 1.5 already released, any update on this release?

Contributor Author

weiqingy commented Jun 12, 2021

@rohitverma02 sorry for the late reply. Will update here about the release plan later. Thanks.

ayakkala1 commented Oct 14, 2022

@weiqingy Is this still being worked on?

hoprocker commented Feb 11, 2023

Hey all, this sounds like a great idea. Imho adding another deployment target (especially one as popular as Kubernetes) would broaden Samza's general adoption appeal. Is this still being targeted for a release, maybe 1.9 at this point?

NoahStapp commented Feb 23, 2023

@weiqingy My team is very interested in this change, as we are looking to move our Samza application to Kubernetes. Is this still being actively worked on? We're interested in helping contribute to this feature, or continuing development on it if this is no longer being worked on.

Contributor

nickpan47 commented Feb 25, 2023 via email

Hi, Noah, Happy to learn that you are interested in this feature. Unfortunately, due to business priority changes, we don't plan to work on that in the near future. I would encourage your team to pick it up if it fits your interests. Regards!

-Yi

On Thu, Feb 23, 2023 at 2:24 PM Noah Stapp ***@***.***> wrote: @weiqingy <https://github.com/weiqingy> My team is very interested in this change, as we are looking to move our Samza application to Kubernetes. Is this still being actively worked on? We're interested in helping contribute to this feature, or continuing development on it if this is no longer being worked on. — Reply to this email directly, view it on GitHub <#1197 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ACL6SBWFE44RPJ4OW36SMFLWY7PR5ANCNFSM4JCOAWAQ> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

Contributor Author

weiqingy commented Feb 25, 2023 •

edited

Loading

Hey @ayakkala1 @hoprocker @NoahStapp Thanks for reviewing and commenting! I haven't got a chance to update PR against the master branch yet, and there are many new changes since the last commit of the PR. Welcome to contribute to this feature and pick it up to catch up your team's timeline.

Best regards,
Weiqing

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet