You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by "baiyangtx (via GitHub)" <gi...@apache.org> on 2024/04/15 03:24:44 UTC
Re: [PR] [AMORO-1958] Implement KubernetesOptimizerContainer to support running Optimizer natively on K8S [amoro]
baiyangtx commented on code in PR #2695:
URL: https://github.com/apache/amoro/pull/2695#discussion_r1565131260
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ Map<String, String> containerProperties1 = getContainerProperties();
+ // start k8s job using k8s client
+ String kubeConfigPath = checkAndGetProperty(containerProperties1, KUBE_CONFIG_PATH);
+ Config config = Config.fromKubeconfig(getKubeConfigContent(kubeConfigPath));
+ this.client = new KubernetesClientBuilder().withConfig(config).build();
+ super.init(name, containerProperties);
+ }
+
+ public 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);
Review Comment:
```suggestion
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);
```
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ Map<String, String> containerProperties1 = getContainerProperties();
+ // start k8s job using k8s client
+ String kubeConfigPath = checkAndGetProperty(containerProperties1, KUBE_CONFIG_PATH);
+ Config config = Config.fromKubeconfig(getKubeConfigContent(kubeConfigPath));
+ this.client = new KubernetesClientBuilder().withConfig(config).build();
+ super.init(name, containerProperties);
+ }
+
+ public 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);
+ }
+ }
+
+ @Override
+ public void requestResource(Resource resource) {
+ // generate pod start args
+ long memoryPerThread =
+ Long.parseLong(checkAndGetProperty(resource.getProperties(), 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 = resource.getProperties().getOrDefault(NAMESPACE, "default");
+ String image = checkAndGetProperty(resource.getProperties(), IMAGE);
+ String cpu = checkAndGetProperty(resource.getProperties(), CPU_PROPERTY);
Review Comment:
Is CPU requests/limits a must?
I think we could ask for a bit more lenient and make it optional. This will provide more flexibility in some test scenarios
##########
ams/dist/src/main/arctic-bin/conf/config.yaml:
##########
@@ -135,8 +135,20 @@ containers:
container-impl: com.netease.arctic.server.manager.LocalOptimizerContainer
properties:
export.JAVA_HOME: "/opt/java" # JDK environment
+ - name: KubernetesContainer
+ container-impl: com.netease.arctic.optimizer.KubernetesOptimizerContainer
+ properties:
+ kube-config-path: ~/.kube/config
+ image: arctic163/amoro:0.6
Review Comment:
KubernetesOptimizerContainer is not intended as a container which is enabled by default.
You can enable it in Helm charts.
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ Map<String, String> containerProperties1 = getContainerProperties();
Review Comment:
Bad local field name. you can use `containerProperties` directly.
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ Map<String, String> containerProperties1 = getContainerProperties();
+ // start k8s job using k8s client
+ String kubeConfigPath = checkAndGetProperty(containerProperties1, KUBE_CONFIG_PATH);
+ Config config = Config.fromKubeconfig(getKubeConfigContent(kubeConfigPath));
+ this.client = new KubernetesClientBuilder().withConfig(config).build();
+ super.init(name, containerProperties);
Review Comment:
Call super.init mulit times.
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
Review Comment:
I think “amoro-optimizer" is a better prefix.
There are some display issues if resource name is too long when using kubctl.
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ Map<String, String> containerProperties1 = getContainerProperties();
+ // start k8s job using k8s client
+ String kubeConfigPath = checkAndGetProperty(containerProperties1, KUBE_CONFIG_PATH);
+ Config config = Config.fromKubeconfig(getKubeConfigContent(kubeConfigPath));
+ this.client = new KubernetesClientBuilder().withConfig(config).build();
+ super.init(name, containerProperties);
+ }
+
+ public 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);
+ }
+ }
+
+ @Override
+ public void requestResource(Resource resource) {
+ // generate pod start args
+ long memoryPerThread =
+ Long.parseLong(checkAndGetProperty(resource.getProperties(), 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 = resource.getProperties().getOrDefault(NAMESPACE, "default");
+ String image = checkAndGetProperty(resource.getProperties(), IMAGE);
+ String cpu = checkAndGetProperty(resource.getProperties(), CPU_PROPERTY);
+
+ String resourceId = resource.getResourceId();
+ String groupName = resource.getGroupName();
+
+ Deployment deployment =
+ new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(NAME + "-" + resourceId)
+ .endMetadata()
Review Comment:
I think that we should support to assign a `K8S Namespace`.
In a production scenario, the namespace of optimizer is usually different from AMS.
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ Map<String, String> containerProperties1 = getContainerProperties();
+ // start k8s job using k8s client
+ String kubeConfigPath = checkAndGetProperty(containerProperties1, KUBE_CONFIG_PATH);
+ Config config = Config.fromKubeconfig(getKubeConfigContent(kubeConfigPath));
+ this.client = new KubernetesClientBuilder().withConfig(config).build();
+ super.init(name, containerProperties);
+ }
+
+ public 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);
+ }
+ }
+
+ @Override
+ public void requestResource(Resource resource) {
+ // generate pod start args
+ long memoryPerThread =
+ Long.parseLong(checkAndGetProperty(resource.getProperties(), MEMORY_PROPERTY));
+ long memory = memoryPerThread * resource.getThreadCount();
Review Comment:
You shoud implement all of those logic in the `protected Map<String, String> doScaleOut(Resource resource)`
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ Map<String, String> containerProperties1 = getContainerProperties();
+ // start k8s job using k8s client
+ String kubeConfigPath = checkAndGetProperty(containerProperties1, KUBE_CONFIG_PATH);
+ Config config = Config.fromKubeconfig(getKubeConfigContent(kubeConfigPath));
+ this.client = new KubernetesClientBuilder().withConfig(config).build();
+ super.init(name, containerProperties);
+ }
+
+ public 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);
+ }
+ }
+
+ @Override
+ public void requestResource(Resource resource) {
+ // generate pod start args
+ long memoryPerThread =
+ Long.parseLong(checkAndGetProperty(resource.getProperties(), 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 = resource.getProperties().getOrDefault(NAMESPACE, "default");
+ String image = checkAndGetProperty(resource.getProperties(), IMAGE);
+ String cpu = checkAndGetProperty(resource.getProperties(), CPU_PROPERTY);
+
+ String resourceId = resource.getResourceId();
+ String groupName = resource.getGroupName();
+
+ Deployment deployment =
+ new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(NAME + "-" + resourceId)
+ .endMetadata()
+ .withNewSpec()
+ .withReplicas(1)
+ .withNewTemplate()
+ .withNewMetadata()
+ .addToLabels("app", NAME + "-" + 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(cpu)))
+ .withRequests(
+ ImmutableMap.of(
+ "memory", new Quantity(memory + "Mi"), "cpu", new Quantity(cpu)))
+ .build())
+ .endContainer()
+ .endSpec()
+ .endTemplate()
+ .withNewSelector()
+ .addToMatchLabels("app", NAME + "-" + resourceId)
+ .endSelector()
+ .endSpec()
+ .build();
+ client.apps().deployments().inNamespace(namespace).resource(deployment).create();
+ }
+
+ @Override
+ protected Map<String, String> doScaleOut(Resource resource) {
+ return null;
+ }
+
+ @Override
+ public void releaseOptimizer(Resource resource) {
+ String resourceId = resource.getResourceId();
+ LOG.info("release Kubernetes Optimizer Container {}", resourceId);
+ client
+ .apps()
+ .deployments()
+ .inNamespace(resource.getGroupName())
Review Comment:
resource group name is not `k8s namespace`
##########
ams/server/src/main/java/com/netease/arctic/server/manager/KubernetesOptimizerContainer.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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 com.netease.arctic.server.manager;
+
+import com.netease.arctic.api.resource.Resource;
+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.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.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_PROPERTY = "cpu";
+ public static final String NAMESPACE = "namespace";
+ public static final String IMAGE = "image";
+ public static final String KUBE_CONFIG_PATH = "kube-config-path";
+
+ public static final String NAME = "amoro-kubernetes-optimizer";
+
+ private KubernetesClient client;
+
+ @Override
+ public void init(String name, Map<String, String> containerProperties) {
+ super.init(name, containerProperties);
+ Map<String, String> containerProperties1 = getContainerProperties();
+ // start k8s job using k8s client
+ String kubeConfigPath = checkAndGetProperty(containerProperties1, KUBE_CONFIG_PATH);
+ Config config = Config.fromKubeconfig(getKubeConfigContent(kubeConfigPath));
+ this.client = new KubernetesClientBuilder().withConfig(config).build();
+ super.init(name, containerProperties);
+ }
+
+ public 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);
+ }
+ }
+
+ @Override
+ public void requestResource(Resource resource) {
+ // generate pod start args
+ long memoryPerThread =
+ Long.parseLong(checkAndGetProperty(resource.getProperties(), 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 = resource.getProperties().getOrDefault(NAMESPACE, "default");
+ String image = checkAndGetProperty(resource.getProperties(), IMAGE);
+ String cpu = checkAndGetProperty(resource.getProperties(), CPU_PROPERTY);
+
+ String resourceId = resource.getResourceId();
+ String groupName = resource.getGroupName();
+
+ Deployment deployment =
+ new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(NAME + "-" + resourceId)
+ .endMetadata()
+ .withNewSpec()
+ .withReplicas(1)
+ .withNewTemplate()
+ .withNewMetadata()
+ .addToLabels("app", NAME + "-" + 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(cpu)))
+ .withRequests(
+ ImmutableMap.of(
+ "memory", new Quantity(memory + "Mi"), "cpu", new Quantity(cpu)))
+ .build())
+ .endContainer()
+ .endSpec()
+ .endTemplate()
+ .withNewSelector()
+ .addToMatchLabels("app", NAME + "-" + resourceId)
+ .endSelector()
+ .endSpec()
+ .build();
+ client.apps().deployments().inNamespace(namespace).resource(deployment).create();
+ }
+
+ @Override
+ protected Map<String, String> doScaleOut(Resource resource) {
+ return null;
+ }
+
+ @Override
+ public void releaseOptimizer(Resource resource) {
+ String resourceId = resource.getResourceId();
+ LOG.info("release Kubernetes Optimizer Container {}", resourceId);
+ client
+ .apps()
+ .deployments()
+ .inNamespace(resource.getGroupName())
+ .withName(NAME + "-" + resourceId)
Review Comment:
I suggest that you store the namespace and deployment's name information in the `doScaleOut`'s return and read it from resource properties during the relesae optimzier, so that there is better compatibility in future code refactoring.
##########
docs/admin-guides/managing-optimizers.md:
##########
@@ -30,6 +30,19 @@ containers:
export.JAVA_HOME: "/opt/java" # JDK environment
```
+### KubernetesOptimizerContainer container
+
+Kubernetes Optimizer Container is a way to start Optimizer On K8s with standalone Optimizer.
+
+```yaml
+containers:
+ - name: KubernetesContainer
+ container-impl: com.netease.arctic.server.manager.KubernetesOptimizerContainer
+ properties:
+ kube-config-path: ～/.kube/config
+ image: arctic163/amoro:0.6
+```
Review Comment:
We should support assign namespace. And user could assign namespace in `$AMS_HOME/conf/config.yaml` or `ResourceGroup`'s properties.
The resource group has high priority.
##########
charts/amoro/templates/_optimizer.tpl:
##########
@@ -84,3 +84,13 @@ properties:
{{- end -}}
{{- end -}}
+{{- define "amoro.optimizer.container.k8s" -}}
+container-impl: com.netease.arctic.server.manager.KubernetesOptimizerContainer
+properties:
+ kube-config-path: "～/.kube/config"
+ image: arctic163/amoro:0.6
Review Comment:
Template the image information from `values.yml`
Or you can implement Helm charts in the other PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@amoro.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org