Skip to content

Commit 381a683

Browse files
jian.hjian-he
jian.h
authored and
jian-he
committed
Support Samza on Kubernetes. Contributed by Weiqing Yang &Jian He
1 parent e7361bb commit 381a683

17 files changed

Lines changed: 1147 additions & 29 deletions

File tree

‎build.gradle‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -568,6 +568,27 @@ project(":samza-yarn_$scalaSuffix") {
568568
jar.dependsOn("lesscss")
569569
}
570570

571+
project(":samza-kubernetes_$scalaSuffix") {
572+
apply plugin: 'java'
573+
574+
dependencies {
575+
compile project(':samza-api')
576+
compile project(":samza-core_$scalaSuffix")
577+
compile "org.codehaus.jackson:jackson-core-asl:1.9.7"
578+
compile group: 'io.fabric8', name: 'kubernetes-client', version: kubernetesJavaClientVersion
579+
testCompile "junit:junit:$junitVersion"
580+
testCompile "org.mockito:mockito-core:$mockitoVersion"
581+
}
582+
583+
tasks.create(name: "releaseKubeTarGz", dependsOn: configurations.archives.artifacts, type: Tar) {
584+
into "samza-kubernetes-${version}"
585+
compression = Compression.GZIP
586+
from(configurations.runtime) { into("lib/") }
587+
from(configurations.archives.artifacts.files) { into("lib/") }
588+
duplicatesStrategy 'exclude'
589+
}
590+
}
591+
571592
project(":samza-shell") {
572593
apply plugin: 'java'
573594

‎gradle/dependency-versions.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,4 +53,5 @@
5353
jnaVersion = "4.5.1"
5454
couchbaseClientVersion = "2.7.2"
5555
couchbaseMockVersion = "1.5.22"
56+
kubernetesJavaClientVersion = "4.1.3"
5657
}

‎gradlew.bat‎

Lines changed: 84 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

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

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ public class ContainerProcessManager implements ClusterResourceManager.Callback
9393
* The Allocator matches requests to resources and executes processes.
9494
*/
9595
private final ContainerAllocator containerAllocator;
96-
private final Thread allocatorThread;
96+
private Thread allocatorThread = null;
9797

9898
// The StandbyContainerManager manages standby-aware allocation and failover of containers
9999
private final Optional<StandbyContainerManager> standbyContainerManager;
@@ -166,8 +166,10 @@ public ContainerProcessManager(Config config, SamzaApplicationState state, Metri
166166
}
167167

168168
this.containerAllocator = new ContainerAllocator(this.clusterResourceManager, config, state, hostAffinityEnabled, this.standbyContainerManager);
169-
this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
170-
LOG.info("Finished container process manager initialization.");
169+
if (shouldStartAllocateThread()) {
170+
this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
171+
}
172+
LOG.info("finished initialization of samza task manager");
171173
}
172174

173175
@VisibleForTesting
@@ -184,14 +186,23 @@ public ContainerProcessManager(Config config, SamzaApplicationState state, Metri
184186

185187
this.clusterResourceManager = resourceManager;
186188
this.standbyContainerManager = Optional.empty();
189+
187190
this.diagnosticsManager = Option.empty();
188191
this.containerAllocator = allocator.orElseGet(
189192
() -> new ContainerAllocator(this.clusterResourceManager, clusterManagerConfig, state,
190193
hostAffinityEnabled, this.standbyContainerManager));
191-
this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
194+
if (shouldStartAllocateThread()) {
195+
this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
196+
}
192197
LOG.info("Finished container process manager initialization");
193198
}
194199

200+
// In Kubernetes, the pod will be started by kubelet automatically once it is allocated, it does not need a
201+
// separate thread to keep polling the allocated resources to start the container.
202+
public boolean shouldStartAllocateThread() {
203+
return !clusterResourceManager.getClass().getSimpleName().equals("KubeClusterResourceManager");
204+
}
205+
195206
public boolean shouldShutdown() {
196207
LOG.debug("ContainerProcessManager state: Completed containers: {}, Configured containers: {}, Are there too many failed containers: {}, Is allocator thread alive: {}",
197208
state.completedProcessors.get(), state.processorCount, jobFailureCriteriaMet ? "yes" : "no", allocatorThread.isAlive() ? "yes" : "no");
@@ -237,7 +248,9 @@ public void start() {
237248

238249
// Start container allocator thread
239250
LOG.info("Starting the container allocator thread");
240-
allocatorThread.start();
251+
if (allocatorThread != null) {
252+
allocatorThread.start();
253+
}
241254
LOG.info("Starting the container process manager");
242255
}
243256

@@ -246,12 +259,15 @@ public void stop() {
246259

247260
// Shutdown allocator thread
248261
containerAllocator.stop();
249-
try {
262+
if (allocatorThread != null) {
263+
264+
try {
250265
allocatorThread.join();
251266
LOG.info("Stopped container allocator");
252267
} catch (InterruptedException ie) {
253-
LOG.error("Allocator thread join threw an interrupted exception", ie);
254-
Thread.currentThread().interrupt();
268+
LOG.error("Allocator thread join threw an interrupted exception", ie);
269+
Thread.currentThread().interrupt();
270+
}
255271
}
256272

257273
if (diagnosticsManager.isDefined()) {

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

Lines changed: 9 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -23,34 +23,22 @@ import java.util
2323
import java.util.concurrent.atomic.AtomicReference
2424

2525
import org.apache.samza.{Partition, SamzaException}
26-
import org.apache.samza.config._
27-
import org.apache.samza.config.Config
28-
import org.apache.samza.container.grouper.stream.SSPGrouperProxy
29-
import org.apache.samza.container.grouper.stream.SystemStreamPartitionGrouperFactory
26+
import org.apache.samza.config.{Config, _}
27+
import org.apache.samza.container.grouper.stream.{SSPGrouperProxy, SystemStreamPartitionGrouperFactory}
3028
import org.apache.samza.container.grouper.task._
31-
import org.apache.samza.coordinator.metadatastore.NamespaceAwareCoordinatorStreamStore
32-
import org.apache.samza.coordinator.stream.messages.SetTaskContainerMapping
33-
import org.apache.samza.coordinator.stream.messages.SetTaskModeMapping
34-
import org.apache.samza.coordinator.stream.messages.SetTaskPartitionMapping
35-
import org.apache.samza.container.LocalityManager
36-
import org.apache.samza.container.TaskName
37-
import org.apache.samza.coordinator.metadatastore.CoordinatorStreamStore
38-
import org.apache.samza.coordinator.server.HttpServer
39-
import org.apache.samza.coordinator.server.JobServlet
40-
import org.apache.samza.coordinator.stream.messages.SetContainerHostMapping
41-
import org.apache.samza.job.model.ContainerModel
42-
import org.apache.samza.job.model.JobModel
43-
import org.apache.samza.job.model.TaskMode
44-
import org.apache.samza.job.model.TaskModel
45-
import org.apache.samza.metrics.MetricsRegistry
46-
import org.apache.samza.metrics.MetricsRegistryMap
29+
import org.apache.samza.container.{LocalityManager, TaskName}
30+
import org.apache.samza.coordinator.metadatastore.{CoordinatorStreamStore, NamespaceAwareCoordinatorStreamStore}
31+
import org.apache.samza.coordinator.server.{HttpServer, JobServlet}
32+
import org.apache.samza.coordinator.stream.messages.{SetContainerHostMapping, SetTaskContainerMapping, SetTaskModeMapping, SetTaskPartitionMapping}
33+
import org.apache.samza.job.model.{ContainerModel, JobModel, TaskMode, TaskModel}
34+
import org.apache.samza.metrics.{MetricsRegistry, MetricsRegistryMap}
4735
import org.apache.samza.runtime.LocationId
4836
import org.apache.samza.system._
4937
import org.apache.samza.util.ScalaJavaUtil.JavaOptionals
5038
import org.apache.samza.util.{Logging, ReflectionUtil, Util}
5139

52-
import scala.collection.JavaConverters
5340
import scala.collection.JavaConversions._
41+
import scala.collection.JavaConverters
5442
import scala.collection.JavaConverters._
5543

5644
/**

‎samza-core/src/main/scala/org/apache/samza/coordinator/server/HttpServer.scala‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -136,4 +136,14 @@ class HttpServer(
136136
throw new SamzaException("HttpServer is not currently running, so URLs are not available for it.")
137137
}
138138
}
139+
140+
def getIpUrl = {
141+
if (running) {
142+
val runningPort = server.getConnectors()(0).asInstanceOf[NetworkConnector].getLocalPort()
143+
144+
new URL("http://" + Util.getLocalHost.getHostAddress + ":" + runningPort + rootPath)
145+
} else {
146+
throw new SamzaException("HttpServer is not currently running, so URLs are not available for it.")
147+
}
148+
}
139149
}

‎samza-core/src/main/scala/org/apache/samza/util/HttpUtil.scala‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ object HttpUtil {
5555
(exception, loop) => {
5656
exception match {
5757
case ioe: IOException => {
58+
error(ioe)
5859
warn("Error getting response from Job coordinator server. received IOException: %s. Retrying..." format ioe.getClass)
5960
httpConn = getHttpConnection(url, timeout)
6061
}

‎samza-kubernetes/src/docker/dockerfiles/Dockerfile‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
#
2+
# Licensed to the Apache Software Foundation (ASF) under one
3+
# or more contributor license agreements. See the NOTICE file
4+
# distributed with this work for additional information
5+
# regarding copyright ownership. The ASF licenses this file
6+
# to you under the Apache License, Version 2.0 (the
7+
# "License"); you may not use this file except in compliance
8+
# with the License. You may obtain a copy of the License at
9+
#
10+
# http://www.apache.org/licenses/LICENSE-2.0
11+
#
12+
# Unless required by applicable law or agreed to in writing,
13+
# software distributed under the License is distributed on an
14+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
# KIND, either express or implied. See the License for the
16+
# specific language governing permissions and limitations
17+
# under the License.
18+
19+
# samzaJarsFolder includes all the Samza jars (you needs to make sure all the samza jars are there.)
20+
# You can build Samza image by:
21+
# docker build -t dockerHubAccount/samza:versionNumber .
22+
# Then Samza user can use the Samza image as base image to build their application image.
23+
#
24+
25+
FROM ubuntu:latest
26+
27+
RUN apt-get update -y && apt-get upgrade -y && apt-get install -y openjdk-8-jdk
28+
29+
ENV JAVA_HOME /usr/lib/jvm/java-8-openjdk-amd64
30+
ENV PATH $PATH:$JAVA_HOME/bin
31+
32+
RUN mkdir -p /opt/samza
33+
WORKDIR /opt/samza/
34+
COPY samzaJarsFolder/ /opt/samza/

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

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.samza.config;
21+
22+
public class KubeConfig {
23+
24+
// the image name of samza
25+
public static final String APP_IMAGE = "kube.app.image";
26+
public static final String DEFAULT_IMAGE = "weiqingyang/samza:v0";
27+
28+
// The directory path inside which the log will be stored.
29+
public static final String SAMZA_MOUNT_DIR = "kube.app.pod.mnt.path";
30+
public static final String K8S_API_NAMESPACE = "kube.app.namespace";
31+
public static final String STREAM_PROCESSOR_CONTAINER_NAME_PREFIX = "sp";
32+
public static final String JC_CONTAINER_NAME_PREFIX = "jc";
33+
public static final String POD_RESTART_POLICY = "OnFailure";
34+
public static final String JC_POD_NAME_FORMAT = "%s-%s-%s"; // jc-appName-appId
35+
public static final String TASK_POD_NAME_FORMAT = "%s-%s-%s-%s"; // sp-appName-appId-containerId
36+
37+
// Environment variable
38+
public static final String COORDINATOR_POD_NAME = "COORDINATOR_POD_NAME";
39+
public static final String AZURE_REMOTE_VOLUME_ENABLED = "kube.app.volume.azure.file-share.enabled";
40+
public static final String AZURE_SECRET = "kube.app.volume.azure-secret";
41+
public static final String AZURE_FILESHARE = "kube.app.volume.azure.file-share";
42+
43+
private Config config;
44+
public KubeConfig(Config config) {
45+
this.config = config;
46+
}
47+
48+
public static KubeConfig validate(Config config) throws ConfigException {
49+
KubeConfig kc = new KubeConfig(config);
50+
kc.validate();
51+
return kc;
52+
}
53+
54+
private void validate() throws ConfigException {
55+
// TODO
56+
}
57+
}

‎samza-kubernetes/src/main/java/org/apache/samza/job/kubernetes/KubeClientFactory.java‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.samza.job.kubernetes;
21+
22+
import io.fabric8.kubernetes.client.Config;
23+
import io.fabric8.kubernetes.client.ConfigBuilder;
24+
import io.fabric8.kubernetes.client.DefaultKubernetesClient;
25+
import io.fabric8.kubernetes.client.KubernetesClient;
26+
import org.slf4j.Logger;
27+
import org.slf4j.LoggerFactory;
28+
29+
30+
public class KubeClientFactory {
31+
private static final Logger LOG = LoggerFactory.getLogger(KubeClientFactory.class);
32+
33+
public static KubernetesClient create() {
34+
ConfigBuilder builder = new ConfigBuilder();
35+
Config config = builder.build();
36+
KubernetesClient client = new DefaultKubernetesClient(config);
37+
LOG.info("Kubernetes client created. ");
38+
return client;
39+
}
40+
}

0 commit comments

Comments
 (0)