This is an automated email from the ASF dual-hosted git repository.
baiyangtx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 6bd2416e6 [AMORO-1958] Implement KubernetesOptimizerContainer to
support running Optimizer natively on K8S (#2695)
6bd2416e6 is described below
commit 6bd2416e6c7d95c7d7d2c4303c7c44fb38bb780b
Author: ConradJam <[email protected]>
AuthorDate: Mon Apr 29 21:33:51 2024 +0800
[AMORO-1958] Implement KubernetesOptimizerContainer to support running
Optimizer natively on K8S (#2695)
* [AMORO-1958] Implement KubernetesOptimizerContainer to support running
Optimizer natively on K8S
* fix override
---
amoro-ams/amoro-ams-server/pom.xml | 5 +
.../manager/KubernetesOptimizerContainer.java | 163 +++++++++++++++++++++
.../server/manager/LocalOptimizerContainer.java | 2 +-
...org.apache.amoro.api.resource.ResourceContainer | 3 +-
.../dist/src/main/arctic-bin/bin/optimizer.sh | 63 +++++++-
.../dist/src/main/arctic-bin/conf/config.yaml | 8 +
charts/amoro/templates/_optimizer.tpl | 8 +
charts/amoro/templates/amoro-configmap.yaml | 4 +
charts/amoro/values.yaml | 7 +
docker/amoro/entrypoint.sh | 2 +-
docs/admin-guides/managing-optimizers.md | 28 ++++
pom.xml | 8 +
12 files changed, 293 insertions(+), 8 deletions(-)
diff --git a/amoro-ams/amoro-ams-server/pom.xml
b/amoro-ams/amoro-ams-server/pom.xml
index e88a6f908..481fc72c8 100644
--- a/amoro-ams/amoro-ams-server/pom.xml
+++ b/amoro-ams/amoro-ams-server/pom.xml
@@ -58,6 +58,11 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.fabric8</groupId>
+ <artifactId>kubernetes-client</artifactId>
+ </dependency>
+
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/manager/KubernetesOptimizerContainer.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/manager/KubernetesOptimizerContainer.java
new file mode 100644
index 000000000..dd18ec6e2
--- /dev/null
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/manager/KubernetesOptimizerContainer.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.manager;
+
+import io.fabric8.kubernetes.api.model.Quantity;
+import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import org.apache.amoro.api.resource.Resource;
+import org.apache.commons.io.IOUtils;
+import org.apache.curator.shaded.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Map;
+
+/** Kubernetes Optimizer Container with Standalone Optimizer */
+public class KubernetesOptimizerContainer extends AbstractResourceContainer {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesOptimizerContainer.class);
+
+ public static final String MEMORY_PROPERTY = "memory";
+ public static final String CPU_FACTOR_PROPERTY = "cpu.factor";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ private static final String NAME_PREFIX = "amoro-optimizer-";
+
+ private static final String KUBERNETES_NAME_PROPERTIES = "name";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ // start k8s job using k8s client
+ String kubeConfigPath = checkAndGetProperty(containerProperties,
KUBE_CONFIG_PATH);
+ Config config =
Config.fromKubeconfig(getKubeConfigContent(kubeConfigPath));
+ this.client = new KubernetesClientBuilder().withConfig(config).build();
+ }
+
+ @Override
+ protected Map<String, String> doScaleOut(Resource resource) {
+ Map<String, String> groupProperties = Maps.newHashMap();
+ groupProperties.putAll(getContainerProperties());
+ groupProperties.putAll(resource.getProperties());
+
+ // generate pod start args
+ long memoryPerThread = Long.parseLong(checkAndGetProperty(groupProperties,
MEMORY_PROPERTY));
+ long memory = memoryPerThread * resource.getThreadCount();
+ // point at amoro home in docker image
+ String startUpArgs =
+ String.format(
+ "/entrypoint.sh optimizer %s %s",
+ memory, super.buildOptimizerStartupArgsString(resource));
+ LOG.info("Starting k8s optimizer using k8s client with start command :
{}", startUpArgs);
+
+ String namespace = groupProperties.getOrDefault(NAMESPACE, "default");
+ String image = checkAndGetProperty(groupProperties, IMAGE);
+ String cpuLimitFactorString =
groupProperties.getOrDefault(CPU_FACTOR_PROPERTY, "1.0");
+ double cpuLimitFactor = Double.parseDouble(cpuLimitFactorString);
+ int cpuLimit = (int) (Math.ceil(cpuLimitFactor *
resource.getThreadCount()));
+
+ String resourceId = resource.getResourceId();
+ String groupName = resource.getGroupName();
+ String kubernetesName = NAME_PREFIX + resourceId;
+ Deployment deployment =
+ new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(NAME_PREFIX + resourceId)
+ .endMetadata()
+ .withNewSpec()
+ .withReplicas(1)
+ .withNewTemplate()
+ .withNewMetadata()
+ .addToLabels("app", NAME_PREFIX + resourceId)
+ .addToLabels("AmoroOptimizerGroup", groupName)
+ .addToLabels("AmoroResourceId", resourceId)
+ .endMetadata()
+ .withNewSpec()
+ .addNewContainer()
+ .withName("optimizer")
+ .withImage(image)
+ .withCommand("sh", "-c", startUpArgs)
+ .withResources(
+ new ResourceRequirementsBuilder()
+ .withLimits(
+ ImmutableMap.of(
+ "memory",
+ new Quantity(memory + "Mi"),
+ "cpu",
+ new Quantity(cpuLimit + "")))
+ .withRequests(
+ ImmutableMap.of(
+ "memory",
+ new Quantity(memory + "Mi"),
+ "cpu",
+ new Quantity(cpuLimit + "")))
+ .build())
+ .endContainer()
+ .endSpec()
+ .endTemplate()
+ .withNewSelector()
+ .addToMatchLabels("app", NAME_PREFIX + resourceId)
+ .endSelector()
+ .endSpec()
+ .build();
+
client.apps().deployments().inNamespace(namespace).resource(deployment).create();
+ Map<String, String> startupProperties = Maps.newHashMap();
+ startupProperties.put(NAMESPACE, namespace);
+ startupProperties.put(KUBERNETES_NAME_PROPERTIES, kubernetesName);
+ return startupProperties;
+ }
+
+ @Override
+ public void releaseOptimizer(Resource resource) {
+ String resourceId = resource.getResourceId();
+ LOG.info("release Kubernetes Optimizer Container {}", resourceId);
+ String namespace = resource.getProperties().get(NAMESPACE);
+ String name = resource.getProperties().get(KUBERNETES_NAME_PROPERTIES);
+ client.apps().deployments().inNamespace(namespace).withName(name).delete();
+ }
+
+ private static String checkAndGetProperty(Map<String, String> properties,
String key) {
+ Preconditions.checkState(
+ properties != null && properties.containsKey(key), "Cannot find %s in
properties", key);
+ return properties.get(key);
+ }
+
+ private String getKubeConfigContent(String path) {
+ try {
+ return IOUtils.toString(Files.newInputStream(Paths.get(path)),
StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/manager/LocalOptimizerContainer.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/manager/LocalOptimizerContainer.java
index 737bbc2b9..2d40e6d36 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/manager/LocalOptimizerContainer.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/manager/LocalOptimizerContainer.java
@@ -55,7 +55,7 @@ public class LocalOptimizerContainer extends
AbstractResourceContainer {
long memoryPerThread =
Long.parseLong(resource.getRequiredProperty(JOB_MEMORY_PROPERTY));
long memory = memoryPerThread * resource.getThreadCount();
return String.format(
- "%s/bin/optimizer.sh %s %s",
+ "%s/bin/optimizer.sh start %s %s",
amsHome, memory, super.buildOptimizerStartupArgsString(resource));
}
diff --git
a/amoro-ams/amoro-ams-server/src/main/resources/META-INF/services/org.apache.amoro.api.resource.ResourceContainer
b/amoro-ams/amoro-ams-server/src/main/resources/META-INF/services/org.apache.amoro.api.resource.ResourceContainer
index d80e47ea6..19a63d196 100644
---
a/amoro-ams/amoro-ams-server/src/main/resources/META-INF/services/org.apache.amoro.api.resource.ResourceContainer
+++
b/amoro-ams/amoro-ams-server/src/main/resources/META-INF/services/org.apache.amoro.api.resource.ResourceContainer
@@ -18,4 +18,5 @@
org.apache.amoro.server.manager.SparkOptimizerContainer
org.apache.amoro.server.manager.FlinkOptimizerContainer
-org.apache.amoro.server.manager.LocalOptimizerContainer
\ No newline at end of file
+org.apache.amoro.server.manager.LocalOptimizerContainer
+org.apache.amoro.server.manager.KubernetesOptimizerContainer
diff --git a/amoro-ams/dist/src/main/arctic-bin/bin/optimizer.sh
b/amoro-ams/dist/src/main/arctic-bin/bin/optimizer.sh
index 2d025f2e5..5d2af0f76 100755
--- a/amoro-ams/dist/src/main/arctic-bin/bin/optimizer.sh
+++ b/amoro-ams/dist/src/main/arctic-bin/bin/optimizer.sh
@@ -23,12 +23,12 @@ CURRENT_DIR="$( cd "$(dirname "$0")" ; pwd -P )"
LIB_PATH=$AMORO_HOME/lib
export CLASSPATH=$AMORO_CONF_DIR/optimize:$LIB_PATH/:$(find $LIB_PATH/ -type f
-name "*.jar" | paste -sd':' -)
-if [ -z $(find $LIB_PATH/ -type f -name "*.jar" | paste -sd':' -) ]; then
+if [ -z "$(find $LIB_PATH/ -type f -name "*.jar" | paste -sd':' -)" ]; then
echo "Launching the localOptimize task lacks relevant jars, please check" >&2
exit 1
fi
-ARGS=${@:2}
+ARGS=${@:3}
if [ -z "$OPTIMIZER_LOG_DIR_NAME" ]; then
OPTIMIZER_LOG_DIR=$AMORO_LOG_DIR/optimizer-$(date +%s)
@@ -37,14 +37,67 @@ else
fi
STDERR_LOG=${OPTIMIZER_LOG_DIR}/optimizer.log.err
-if [ ! -f "${OPTIMIZER_LOG_DIR}" ]; then
+if [ ! -d "${OPTIMIZER_LOG_DIR}" ]; then
mkdir -p "${OPTIMIZER_LOG_DIR}"
fi
if [ ! -f $STDERR_LOG ];then
touch $STDERR_LOG
fi
-JAVA_OPTS="-Xmx$1m -Dlog.home=${OPTIMIZER_LOG_DIR}"
+JAVA_OPTS="-Xmx$2m -Dlog.home=${OPTIMIZER_LOG_DIR}"
RUN_SERVER="org.apache.amoro.optimizer.standalone.StandaloneOptimizer"
CMDS="$JAVA_RUN $JAVA_OPTS $RUN_SERVER $ARGS"
-nohup ${CMDS} >/dev/null 2>${STDERR_LOG} &
+
+start() {
+ nohup $CMDS > /dev/null 2>"$STDERR_LOG" &
+ echo "Amoro Optimizer started in background with PID $!"
+}
+
+start-foreground() {
+ $CMDS
+}
+
+#0:pid bad and proc OK; 1:pid ok and proc bad; 2:pid bad
+function status(){
+ test -e ${PID} || return 2
+ test -n "$(ps -p $(cat ${PID}) -o pid=)" && return 0 || return 1
+}
+
+case "$1" in
+ start)
+ start
+ ;;
+ start-foreground)
+ start-foreground
+ ;;
+ stop)
+ status;
+ if [ $? -ne 0 ]; then
+ echo "proc not running."
+ else
+ echo 'stopping app server.'
+ stop
+ fi
+ ;;
+ status)
+ status;
+ if [ $? -eq 0 ];then
+ echo 'running.'
+ else
+ echo 'not running.'
+ exit 1
+ fi
+ ;;
+ pid)
+ status
+ if [ $? -eq 0 ]; then
+ cat $PID
+ else
+ echo "not running"
+ fi
+ ;;
+ *)
+ echo "Usage: $0 {start|start-foreground|stop|status|pid}
<memory_limit_in_MB> [additional_args]"
+ exit 1
+ ;;
+esac
\ No newline at end of file
diff --git a/amoro-ams/dist/src/main/arctic-bin/conf/config.yaml
b/amoro-ams/dist/src/main/arctic-bin/conf/config.yaml
index 02218a5c9..d6cbb7ff2 100644
--- a/amoro-ams/dist/src/main/arctic-bin/conf/config.yaml
+++ b/amoro-ams/dist/src/main/arctic-bin/conf/config.yaml
@@ -137,6 +137,14 @@ containers:
export.JAVA_HOME: "/opt/java" # JDK environment
#containers:
+
+# - name: KubernetesContainer
+# container-impl:
com.netease.arctic.server.manager.KubernetesOptimizerContainer
+# properties:
+# kube-config-path: ～/.kube/config
+# image: arctic163/amoro:0.6
+# namespace: default
+
# - name: flinkContainer
# container-impl: org.apache.amoro.server.manager.FlinkOptimizerContainer
# properties:
diff --git a/charts/amoro/templates/_optimizer.tpl
b/charts/amoro/templates/_optimizer.tpl
index 48847ed96..6bfb1a2ac 100644
--- a/charts/amoro/templates/_optimizer.tpl
+++ b/charts/amoro/templates/_optimizer.tpl
@@ -84,3 +84,11 @@ properties:
{{- end -}}
{{- end -}}
+{{- define "amoro.optimizer.container.kubernetes" -}}
+container-impl: org.apache.amoro.server.manager.KubernetesOptimizerContainer
+properties:
+ {{- with .Values.optimizer.kubernetes.properties -}}
+ {{- toYaml . | nindent 2 }}
+ {{- end -}}
+{{- end -}}
+
diff --git a/charts/amoro/templates/amoro-configmap.yaml
b/charts/amoro/templates/amoro-configmap.yaml
index bbc02fe0b..e413ed977 100644
--- a/charts/amoro/templates/amoro-configmap.yaml
+++ b/charts/amoro/templates/amoro-configmap.yaml
@@ -113,6 +113,10 @@ data:
- name: {{ .Values.optimizer.local.name | default "local" }}
{{- include "amoro.optimizer.container.local" . | nindent 8 }}
{{ end }}
+ {{- if .Values.optimizer.kubernetes.enabled }}
+ - name: {{ .Values.optimizer.local.name | default "kubernetes" }}
+ {{- include "amoro.optimizer.container.kubernetes" . | nindent 8 }}
+ {{ end }}
{{- with .Values.amoroConf.amoroDefaults }}
diff --git a/charts/amoro/values.yaml b/charts/amoro/values.yaml
index e364980e3..aa01ca4d3 100644
--- a/charts/amoro/values.yaml
+++ b/charts/amoro/values.yaml
@@ -171,6 +171,13 @@ optimizer:
name: ~
# properties for optimizer container.
properties: ~
+ kubernetes:
+ # enable the kubernetes optimizer container
+ enabled: true
+ properties:
+ namespace: "default"
+ kube-config-path: "～/.kube/config"
+ image: "arctic163/amoro:0.6"
extra: []
diff --git a/docker/amoro/entrypoint.sh b/docker/amoro/entrypoint.sh
index 2b857a40e..57d867373 100644
--- a/docker/amoro/entrypoint.sh
+++ b/docker/amoro/entrypoint.sh
@@ -56,7 +56,7 @@ elif [ "$1" == "ams" ]; then
elif [ "$1" == "optimizer" ]; then
args=("${args[@]:1}")
echo "Start Amoro Optimizer"
- exec "$AMORO_HOME/bin/optimizer.sh " "${args[@]}"
+ exec "$AMORO_HOME/bin/optimizer.sh" "start-foreground" "${args[@]}"
fi
# Running command in pass-through mode
diff --git a/docs/admin-guides/managing-optimizers.md
b/docs/admin-guides/managing-optimizers.md
index 982d9db3c..613e70f21 100644
--- a/docs/admin-guides/managing-optimizers.md
+++ b/docs/admin-guides/managing-optimizers.md
@@ -30,6 +30,34 @@ containers:
export.JAVA_HOME: "/opt/java" # JDK environment
```
+### Kubernetes container
+
+Kubernetes Optimizer Container is a way to start Optimizer On K8s with
standalone Optimizer.
+To use flink container, you need to add a new container configuration.
+with container-impl as
`org.apache.amoro.server.manager.KubernetesOptimizerContainer`
+
+KubernetesOptimizerContainer support the following properties:
+
+
+| Property Name | Required | Default Value | Description
|
+|---------------------------|----------|---------------|---------------------------------------------------------------------------------------------------------------|
+| kube-config-path | true | N/A | Kubernetes config
location.
|
+| image | true | N/A | Optimizer Image name
|
+| namespace | false | "default" | The namespace of
optimizer to deploy
|
+| ams-optimizing-uri | false | N/A | uri of AMS thrift
self-optimizing endpoint. This could be used if the ams.server-expose-host is
not available |
+| cpu.factor | false | "1.0" | Cpu factor when
request kubernetes resource. Default 1 Cpu pre thread
|
+| memory | true | N/A | Memory usage for pre
thread
|
+
+
+```yaml
+containers:
+ - name: KubernetesContainer
+ container-impl:
org.apache.amoro.server.manager.KubernetesOptimizerContainer
+ properties:
+ kube-config-path: ～/.kube/config
+ image: apache/amoro:0.6
+```
+
### Flink container
Flink container is a way to start Optimizer through Flink jobs. With Flink,
you can easily deploy Optimizer
on yarn clusters or kubernetes clusters to support large-scale data scenarios.
To use flink container,
diff --git a/pom.xml b/pom.xml
index 480f68767..8f90ffc12 100644
--- a/pom.xml
+++ b/pom.xml
@@ -128,6 +128,7 @@
<bitmap.version>1.0.1</bitmap.version>
<prometheus.version>0.16.0</prometheus.version>
<flink.version>1.18.1</flink.version>
+
<fabric8-kubernetes-client.version.version>6.12.1</fabric8-kubernetes-client.version.version>
</properties>
<dependencies>
@@ -793,6 +794,13 @@
<version>${flink.version}</version>
</dependency>
+ <dependency>
+ <groupId>io.fabric8</groupId>
+ <artifactId>kubernetes-client</artifactId>
+ <version>${fabric8-kubernetes-client.version.version}</version>
+ </dependency>
+
+
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
