Skip to content

[fix](cloud) Replace compute group properties from Meta Service #67911

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

Open
deardeng wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[fix](cloud) Replace compute group properties from Meta Service #67911

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 20 additions & 10 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudComputeGroupMeta.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,7 @@ public Cloud.ClusterPolicy toPb() {

@Getter
@Setter
private Map<String, String> properties = new LinkedHashMap<>(ALL_PROPERTIES_DEFAULT_VALUE_MAP);
private volatile Map<String, String> properties = new LinkedHashMap<>(ALL_PROPERTIES_DEFAULT_VALUE_MAP);

public CloudComputeGroupMeta(String id, String name, ComputeTypeEnum type) {
this.id = id;
Expand Down Expand Up @@ -245,28 +245,32 @@ public void checkProperties(Map<String, String> inputProperties) throws DdlExcep
}

public void modifyProperties(Map<String, String> inputProperties) throws DdlException {
String balanceType = inputProperties.get(BALANCE_TYPE);
applyBalanceTypeRule(properties, inputProperties.get(BALANCE_TYPE));
}

// only async_warmup carries a timeout: drop it for other types, fill the default when missing
private static void applyBalanceTypeRule(Map<String, String> target, String balanceType) {
if (balanceType == null) {
return;
}
if (BalanceTypeEnum.WITHOUT_WARMUP.getValue().equals(balanceType)
|| BalanceTypeEnum.SYNC_WARMUP.getValue().equals(balanceType)
|| BalanceTypeEnum.PEER_READ_ASYNC_WARMUP.getValue().equals(balanceType)) {
// delete BALANCE_WARM_UP_TASK_TIMEOUT if exists
properties.remove(BALANCE_WARM_UP_TASK_TIMEOUT);
target.remove(BALANCE_WARM_UP_TASK_TIMEOUT);
} else if (BalanceTypeEnum.ASYNC_WARMUP.getValue().equals(balanceType)) {
// if BALANCE_WARM_UP_TASK_TIMEOUT exists, it has been validated in validateProperty
if (!properties.containsKey(BALANCE_WARM_UP_TASK_TIMEOUT)) {
properties.put(BALANCE_WARM_UP_TASK_TIMEOUT, String.valueOf(DEFAULT_BALANCE_WARM_UP_TASK_TIMEOUT));
if (!target.containsKey(BALANCE_WARM_UP_TASK_TIMEOUT)) {
target.put(BALANCE_WARM_UP_TASK_TIMEOUT, String.valueOf(DEFAULT_BALANCE_WARM_UP_TASK_TIMEOUT));
}
}
}

// set properties, just set in periodic instance status checker
// set properties, just set in periodic instance status checker.
// MS is the source of truth and stores only what was explicitly set, so FE properties =
// FE config defaults overlaid by the MS snapshot: keys removed in MS fall back to defaults.
public void setProperties(Map<String, String> propertiesInMs) {
if (propertiesInMs == null || propertiesInMs.isEmpty()) {
return;
}
Map<String, String> newProperties = new LinkedHashMap<>(ALL_PROPERTIES_DEFAULT_VALUE_MAP);

for (Map.Entry<String, String> entry : propertiesInMs.entrySet()) {
String key = entry.getKey();
Expand All @@ -281,9 +285,15 @@ public void setProperties(Map<String, String> propertiesInMs) {
}

if (value != null && !value.isEmpty()) {
properties.put(key, value);
newProperties.put(key, value);
}
}
applyBalanceTypeRule(newProperties, newProperties.get(BALANCE_TYPE));

if (!newProperties.equals(properties)) {
LOG.info("compute group {} properties changed: {} -> {}", name, properties, newProperties);
properties = newProperties;
}
}

public BalanceTypeEnum getBalanceType() {
Expand Down
40 changes: 2 additions & 38 deletions fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,48 +121,12 @@ private void handleComputeClusters(List<Cloud.ClusterPB> computeClusters) {
+ "it may be wait cluster checker to sync, ignore it",
computeClusterInMs);
} else {
// exist compute group, check properties changed and update if needed
updatePropertiesIfChanged(computeGroupInFe, computeClusterInMs);
// exist compute group, resync properties from the authoritative MS snapshot
computeGroupInFe.setProperties(computeClusterInMs.getPropertiesMap());
}
}
}

/**
* Compare properties between compute cluster in MS and compute group in FE,
* update only the changed key-value pairs to avoid unnecessary updates.
*/
private void updatePropertiesIfChanged(CloudComputeGroupMeta computeGroupInFe, Cloud.ClusterPB computeClusterInMs) {
Map<String, String> propertiesInMs = computeClusterInMs.getPropertiesMap();
Map<String, String> propertiesInFe = computeGroupInFe.getProperties();

if (propertiesInMs == null || propertiesInMs.isEmpty()) {
return;
}
Map<String, String> changedProperties = new HashMap<>();

// Check for changed or new properties
for (Map.Entry<String, String> entry : propertiesInMs.entrySet()) {
String key = entry.getKey();
String valueInMs = entry.getValue();
String valueInFe = propertiesInFe.get(key);

if (valueInFe != null && valueInFe.equalsIgnoreCase(valueInMs)) {
continue;
}
changedProperties.put(key, valueInMs);

LOG.debug("Property changed for compute group {}: {} = {} (was: {})",
computeGroupInFe.getName(), key, valueInMs, valueInFe);
}

// Only update if there are actual changes
if (!changedProperties.isEmpty()) {
LOG.info("Updating properties for compute group {}: {}",
computeGroupInFe.getName(), changedProperties);
computeGroupInFe.setProperties(changedProperties);
}
}

private void categorizeClusters(List<Cloud.ClusterPB> clusters,
List<Cloud.ClusterPB> virtualClusters, List<Cloud.ClusterPB> computeClusters) {
for (Cloud.ClusterPB cluster : clusters) {
Expand Down
8 changes: 7 additions & 1 deletion ...src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterComputeGroupCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.apache.doris.qe.StmtExecutor;

//import org.apache.commons.lang3.StringUtils;
import java.util.LinkedHashMap;
import java.util.Map;

/**
Expand Down Expand Up @@ -87,8 +88,13 @@ public void validate(ConnectContext connectContext) throws UserException {
public void doRun(ConnectContext ctx, StmtExecutor executor) throws Exception {
validate(ctx);
CloudSystemInfoService cloudSys = ((CloudSystemInfoService) Env.getCurrentSystemInfo());
// MS replaces the whole property map, so send current (timeout rule already applied
// by validate) overlaid with the user input, not just the changed keys
Map<String, String> merged = new LinkedHashMap<>(
cloudSys.getComputeGroupByName(computeGroupName).getProperties());
merged.putAll(properties);
// send rpc to ms
cloudSys.alterComputeGroupProperties(computeGroupName, properties);
cloudSys.alterComputeGroupProperties(computeGroupName, merged);
}

@Override
Expand Down
39 changes: 39 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,32 @@ public void testSyncInstanceCreatesVirtualComputeGroup() {
Assertions.assertEquals("standby_cg", virtualComputeGroup.getStandbyComputeGroup());
}

@Test
public void testSyncComputeGroupPropertiesRemovesKeysMissingFromMetaService() {
CloudComputeGroupMeta computeGroup = new CloudComputeGroupMeta(
"compute_cg_id", "compute_cg", CloudComputeGroupMeta.ComputeTypeEnum.COMPUTE);
Map<String, String> initialProperties = new HashMap<>();
initialProperties.put(CloudComputeGroupMeta.BALANCE_TYPE, BalanceTypeEnum.ASYNC_WARMUP.getValue());
initialProperties.put(CloudComputeGroupMeta.BALANCE_WARM_UP_TASK_TIMEOUT, "900");
computeGroup.setProperties(initialProperties);
cloudSystemInfoService.addComputeGroup("compute_cg_id", computeGroup);

Cloud.ClusterPB propertiesInMetaService = Cloud.ClusterPB.newBuilder()
.setClusterId("compute_cg_id")
.setClusterName("compute_cg")
.setType(Cloud.ClusterPB.Type.COMPUTE)
.putProperties(CloudComputeGroupMeta.BALANCE_TYPE, BalanceTypeEnum.WITHOUT_WARMUP.getValue())
.build();
Mockito.doReturn(instanceResponse(propertiesInMetaService))
.when(cloudSystemInfoService).getCloudInstance();

new CloudInstanceStatusChecker(cloudSystemInfoService).runAfterCatalogReady();

Assertions.assertEquals(propertiesInMetaService.getPropertiesMap(), computeGroup.getProperties());
Assertions.assertEquals(CloudComputeGroupMeta.DEFAULT_BALANCE_WARM_UP_TASK_TIMEOUT,
computeGroup.getBalanceWarmUpTaskTimeout());
}

@Test
public void testSyncInstanceCreatesVirtualComputeGroupAndCancelsTableLevelLoadEvent() throws Exception {
databases.add(mockDb("ods", mockTable(1001, "orders")));
Expand Down Expand Up @@ -309,6 +335,19 @@ private Cloud.GetInstanceResponse instanceResponseWithoutVirtualComputeGroup() {
.build();
}

private Cloud.GetInstanceResponse instanceResponse(Cloud.ClusterPB computeGroup) {
return Cloud.GetInstanceResponse.newBuilder()
.setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
.setCode(Cloud.MetaServiceCode.OK)
.setMsg("OK")
.build())
.setInstance(Cloud.InstanceInfoPB.newBuilder()
.setStatus(Cloud.InstanceInfoPB.Status.NORMAL)
.addClusters(computeGroup)
.build())
.build();
}

private Cloud.ClusterPB computeGroup(String computeGroupId, String computeGroupName) {
return Cloud.ClusterPB.newBuilder()
.setClusterId(computeGroupId)
Expand Down
Loading