Skip to content

Commit

Permalink
Any details of the changes
Browse files Browse the repository at this point in the history
  • Loading branch information
@wangyuxiang123
wangyuxiang123 committed Sep 3, 2024
1 parent d0ca3f2 commit 9388031
Show file tree
Hide file tree
Showing 2 changed files with 83 additions and 69 deletions.
88 changes: 58 additions & 30 deletions ...ms-server/src/main/java/org/apache/amoro/server/manager/KubernetesOptimizerContainer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -88,38 +88,18 @@ protected Map<String, String> doScaleOut(Resource resource) {
groupProperties.putAll(getContainerProperties());
groupProperties.putAll(resource.getProperties());

// generate pod start args
long memoryPerThread;
long memory;
//
if (resource.getMemoryMb() > 0) {
memory = resource.getMemoryMb();
} else {
memoryPerThread = Long.parseLong(checkAndGetProperty(groupProperties, MEMORY_PROPERTY));
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 pullPolicy = checkAndGetProperty(groupProperties, PULL_POLICY);
String pullSecrets = groupProperties.getOrDefault(PULL_SECRETS, "");
String cpuLimitFactorString = groupProperties.getOrDefault(CPU_FACTOR_PROPERTY, "1.0");
double cpuLimitFactor = Double.parseDouble(cpuLimitFactorString);
int cpuLimit = (int) (Math.ceil(cpuLimitFactor * resource.getThreadCount()));

Map<String, Object> argsList = generatePodStartArgs(resource, groupProperties);
String image = argsList.get(IMAGE).toString();
String namespace = argsList.get(NAMESPACE).toString();
String pullPolicy = argsList.get(PULL_POLICY).toString();
List<LocalObjectReference> imagePullSecretsList =
Arrays.stream(pullSecrets.split(";"))
.map(secret -> new LocalObjectReferenceBuilder().withName(secret).build())
.collect(Collectors.toList());
(List<LocalObjectReference>) argsList.get(PULL_SECRETS);
int cpuLimit = (int) argsList.get("cpuLimit");
long memory = (long) argsList.get(MEMORY_PROPERTY);
String groupName = argsList.get("groupName").toString();
String resourceId = argsList.get("resourceId").toString();
String startUpArgs = argsList.get("startUpArgs").toString();

String resourceId = resource.getResourceId();
String groupName = resource.getGroupName();
String kubernetesName = NAME_PREFIX + resourceId;
Deployment deployment;

Expand Down Expand Up @@ -158,6 +138,54 @@ protected Map<String, String> doScaleOut(Resource resource) {
return startupProperties;
}

public Map<String, Object> generatePodStartArgs(
Resource resource, Map<String, String> groupProperties) {
long memoryPerThread;
long memory;

if (resource.getMemoryMb() > 0) {
memory = resource.getMemoryMb();
} else {
memoryPerThread = Long.parseLong(checkAndGetProperty(groupProperties, MEMORY_PROPERTY));
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 pullPolicy = checkAndGetProperty(groupProperties, PULL_POLICY);
String pullSecrets = groupProperties.getOrDefault(PULL_SECRETS, "");
String cpuLimitFactorString = groupProperties.getOrDefault(CPU_FACTOR_PROPERTY, "1.0");
double cpuLimitFactor = Double.parseDouble(cpuLimitFactorString);
int cpuLimit = (int) (Math.ceil(cpuLimitFactor * resource.getThreadCount()));

List<LocalObjectReference> imagePullSecretsList =
Arrays.stream(pullSecrets.split(";"))
.map(secret -> new LocalObjectReferenceBuilder().withName(secret).build())
.collect(Collectors.toList());

String resourceId = resource.getResourceId();
String groupName = resource.getGroupName();

Map<String, Object> argsList = Maps.newHashMap();
argsList.put(NAMESPACE, namespace);
argsList.put(IMAGE, image);
argsList.put(PULL_POLICY, pullPolicy);
argsList.put(PULL_SECRETS, imagePullSecretsList);
argsList.put(MEMORY_PROPERTY, memory);
argsList.put("cpuLimit", cpuLimit);
argsList.put("resourceId", resourceId);
argsList.put("groupName", groupName);
argsList.put("startUpArgs", startUpArgs);

return argsList;
}

public Deployment initPodTemplateWithoutConfig(
String image,
String pullPolicy,
Expand Down
64 changes: 25 additions & 39 deletions ...erver/src/test/java/org/apache/amoro/server/manager/TestKubernetesOptimizerContainer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,10 @@
import io.fabric8.kubernetes.api.model.LocalObjectReference;
import io.fabric8.kubernetes.api.model.LocalObjectReferenceBuilder;
import io.fabric8.kubernetes.api.model.PodTemplate;
import io.fabric8.kubernetes.api.model.Quantity;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import org.apache.amoro.api.OptimizerProperties;
import org.apache.amoro.api.resource.Resource;
import org.apache.amoro.api.resource.ResourceType;
import org.apache.amoro.OptimizerProperties;
import org.apache.amoro.resource.Resource;
import org.apache.amoro.resource.ResourceType;
import org.apache.amoro.server.AmoroManagementConf;
import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
Expand Down Expand Up @@ -114,34 +113,17 @@ public void testBuildPodTemplateWithResourceSetMemoryMb() {
.build();
groupProperties.putAll(resource.getProperties());

// generate pod start args
long memoryPerThread;
long memory;

if (resource.getMemoryMb() > 0) {
memory = resource.getMemoryMb();
} else {
memoryPerThread = Long.parseLong(checkAndGetProperty(groupProperties, MEMORY_PROPERTY));
memory = memoryPerThread * resource.getThreadCount();
}
String startUpArgs =
String.format(
"/entrypoint.sh optimizer %s %s",
memory, kubernetesOptimizerContainer.buildOptimizerStartupArgsString(resource));
String image = checkAndGetProperty(groupProperties, IMAGE);
String pullPolicy = checkAndGetProperty(groupProperties, PULL_POLICY);
String pullSecrets = groupProperties.getOrDefault(PULL_SECRETS, "");
String cpuLimitFactorString = groupProperties.getOrDefault(CPU_FACTOR_PROPERTY, "1.0");
double cpuLimitFactor = Double.parseDouble(cpuLimitFactorString);
int cpuLimit = (int) (Math.ceil(cpuLimitFactor * resource.getThreadCount()));

Map<String, Object> argsList =
kubernetesOptimizerContainer.generatePodStartArgs(resource, groupProperties);
String image = argsList.get(IMAGE).toString();
String pullPolicy = argsList.get(PULL_POLICY).toString();
List<LocalObjectReference> imagePullSecretsList =
Arrays.stream(pullSecrets.split(";"))
.map(secret -> new LocalObjectReferenceBuilder().withName(secret).build())
.collect(Collectors.toList());

String resourceId = resource.getResourceId();
String groupName = resource.getGroupName();
(List<LocalObjectReference>) argsList.get(PULL_SECRETS);
int cpuLimit = (int) argsList.get("cpuLimit");
long memory = (long) argsList.get(MEMORY_PROPERTY);
String groupName = argsList.get("groupName").toString();
String resourceId = argsList.get("resourceId").toString();
String startUpArgs = argsList.get("startUpArgs").toString();

Deployment deployment =
kubernetesOptimizerContainer.initPodTemplateFromFrontEnd(
Expand All @@ -157,7 +139,7 @@ public void testBuildPodTemplateWithResourceSetMemoryMb() {

Assert.assertEquals(1, deployment.getSpec().getReplicas().intValue());
Assert.assertNotEquals(
new Quantity("1024Mi"),
"1024Mi",
deployment
.getSpec()
.getTemplate()
Expand All @@ -166,9 +148,10 @@ public void testBuildPodTemplateWithResourceSetMemoryMb() {
.get(0)
.getResources()
.getLimits()
.get("memory"));
.get("memory")
.toString());
Assert.assertEquals(
new Quantity("1025Mi"),
"1025Mi",
deployment
.getSpec()
.getTemplate()
Expand All @@ -177,7 +160,8 @@ public void testBuildPodTemplateWithResourceSetMemoryMb() {
.get(0)
.getResources()
.getLimits()
.get("memory"));
.get("memory")
.toString());
}

@Test
Expand Down Expand Up @@ -263,7 +247,7 @@ public void testBuildPodTemplateConfig() {
"apache/amoro:0.7-SNAPSHOT",
deployment.getSpec().getTemplate().getSpec().getContainers().get(0).getImage());
Assert.assertEquals(
new Quantity("1"),
String.valueOf(cpuLimit),
deployment
.getSpec()
.getTemplate()
Expand All @@ -272,9 +256,10 @@ public void testBuildPodTemplateConfig() {
.get(0)
.getResources()
.getLimits()
.get("cpu"));
.get("cpu")
.toString());
Assert.assertEquals(
new Quantity("1024Mi"),
memory + "Mi",
deployment
.getSpec()
.getTemplate()
Expand All @@ -283,6 +268,7 @@ public void testBuildPodTemplateConfig() {
.get(0)
.getResources()
.getLimits()
.get("memory"));
.get("memory")
.toString());
}
}

0 comments on commit 9388031

Please sign in to comment.