diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java index c8eee36f505a31..c38bc7debc81ae 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudClusterChecker.java @@ -27,6 +27,7 @@ import org.apache.doris.common.Config; import org.apache.doris.common.DdlException; import org.apache.doris.common.UserException; +import org.apache.doris.common.util.DebugPointUtil; import org.apache.doris.common.util.MasterDaemon; import org.apache.doris.ha.FrontendNodeType; import org.apache.doris.metric.MetricRepo; @@ -146,19 +147,22 @@ private void checkToDelCluster(Map remoteClusterIdToPB, Set toDel = new ArrayList<>(finalClusterIdToBackend.getOrDefault(delId, new ArrayList<>())); + // The name index may already belong to a same-name replacement. Use the + // obsolete group's own BE tags, and never skip ID cleanup for a missing name. + String delClusterName = toDel.stream().map(Backend::getCloudClusterName).findFirst() + .orElseGet(() -> cloudSystemInfoService.getClusterNameByClusterId(delId)); + // Name-scoped jobs may already belong to the replacement group. + if (delId.equals(cloudSystemInfoService.getCloudClusterIdByName(delClusterName))) { + ((CloudEnv) Env.getCurrentEnv()).getCacheHotspotMgr().cancelTableFilterJobsForClusterChange( + delClusterName, "system cancel: compute group " + delClusterName + " dropped"); + } cloudSystemInfoService.updateCloudBackends(new ArrayList<>(), toDel); // del clusterName // del clusterID MetricRepo.unregisterCloudMetrics(delId, delClusterName, toDel); - cloudSystemInfoService.dropCluster(delId, delClusterName); + cloudSystemInfoService.removeComputeGroup(delId, delClusterName); } ); } @@ -539,6 +543,11 @@ private void checkCloudFes() { } private void checkCloudBackends() { + if (DebugPointUtil.isEnable("CloudClusterChecker.checkCloudBackends.pause")) { + LOG.info("CloudClusterChecker.checkCloudBackends.pause phase={}", DebugPointUtil.getDebugParamOrDefault( + "CloudClusterChecker.checkCloudBackends.pause", "phase", "")); + return; + } Map> clusterIdToBackend = cloudSystemInfoService.getCloudClusterIdToBackend(false); //rpc to ms, to get mysql user can use cluster_id // NOTE: rpc args all empty, use cluster_unique_id to get a instance's all cluster info. @@ -574,6 +583,8 @@ private void checkCloudBackends() { // clusterID local == remote, diff nodes checkDiffNode(remoteClusterIdToPB, clusterIdToBackend); + cloudSystemInfoService.refreshComputeGroupNames(remoteClusterIdToPB.values()); + // check mem map checkFeNodesMapValid(); } catch (Exception e) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java index cc9d256d566f38..72742911dafbd4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java @@ -26,6 +26,7 @@ import org.apache.doris.common.Config; import org.apache.doris.common.DdlException; import org.apache.doris.common.Pair; +import org.apache.doris.common.util.DebugPointUtil; import org.apache.doris.common.util.MasterDaemon; import org.apache.doris.metric.MetricRepo; import org.apache.doris.nereids.trees.plans.commands.WarmUpClusterCommand; @@ -58,6 +59,12 @@ public CloudInstanceStatusChecker(CloudSystemInfoService cloudSystemInfoService) @Override protected void runAfterCatalogReady() { try { + if (DebugPointUtil.isEnable("CloudInstanceStatusChecker.runAfterCatalogReady.pause")) { + LOG.info("CloudInstanceStatusChecker.runAfterCatalogReady.pause phase={}", + DebugPointUtil.getDebugParamOrDefault( + "CloudInstanceStatusChecker.runAfterCatalogReady.pause", "phase", "")); + return; + } long start = System.currentTimeMillis(); Cloud.GetInstanceResponse response = cloudSystemInfoService.getCloudInstance(); if (!isResponseValid(response)) { @@ -65,6 +72,17 @@ protected void runAfterCatalogReady() { } Cloud.InstanceInfoPB instance = response.getInstance(); + if (DebugPointUtil.isEnable("CloudInstanceStatusChecker.afterGetInstance.pause") + && instance.getClustersList().stream().anyMatch(c -> c.getClusterId().equals( + DebugPointUtil.getDebugParamOrDefault( + "CloudInstanceStatusChecker.afterGetInstance.pause", "cluster_id", "")))) { + LOG.info("CloudInstanceStatusChecker.afterGetInstance.pause phase={}", + DebugPointUtil.getDebugParamOrDefault( + "CloudInstanceStatusChecker.afterGetInstance.pause", "phase", "")); + while (DebugPointUtil.isEnable("CloudInstanceStatusChecker.afterGetInstance.pause")) { + Thread.sleep(100); + } + } cloudSystemInfoService.setInstanceStatus(instance.getStatus()); syncStorageVault(instance); processVirtualClusters(instance.getClustersList()); @@ -107,8 +125,10 @@ private void processVirtualClusters(List clusters) { List computeClusters = new ArrayList<>(); categorizeClusters(clusters, virtualClusters, computeClusters); handleComputeClusters(computeClusters); - handleVirtualClusters(virtualClusters, computeClusters); + List reconciledVirtualClusters = handleVirtualClusters(virtualClusters, computeClusters); + // A rejected update still proves that the group exists in MS, so use the full snapshot for removal. removeObsoleteVirtualGroups(virtualClusters); + cloudSystemInfoService.refreshComputeGroupNames(reconciledVirtualClusters); } private void handleComputeClusters(List computeClusters) { @@ -179,15 +199,23 @@ private void categorizeClusters(List clusters, } } - private void handleVirtualClusters(List virtualGroups, List computeClusters) { + private List handleVirtualClusters( + List virtualGroups, List computeClusters) { + List reconciledGroups = new ArrayList<>(); for (Cloud.ClusterPB virtualGroupInMs : virtualGroups) { CloudComputeGroupMeta virtualGroupInFe = cloudSystemInfoService .getComputeGroupById(virtualGroupInMs.getClusterId()); + boolean reconciled; if (virtualGroupInFe != null) { - handleExistingVirtualComputeGroup(virtualGroupInMs, virtualGroupInFe); + reconciled = handleExistingVirtualComputeGroup(virtualGroupInMs, virtualGroupInFe); } else { - handleNewVirtualComputeGroup(virtualGroupInMs, computeClusters); + reconciled = handleNewVirtualComputeGroup(virtualGroupInMs, computeClusters); + } + // Rejected renames must not publish a name for metadata whose update was rejected. + if (!reconciled) { + continue; } + reconciledGroups.add(virtualGroupInMs); // just fe master gen file cache sync task if (Env.getCurrentEnv().isMaster()) { // get again in fe mem @@ -201,6 +229,7 @@ private void handleVirtualClusters(List virtualGroups, List jobIds) { @@ -349,21 +378,22 @@ private void syncFileCacheTasksForVirtualGroup( } } - private void handleExistingVirtualComputeGroup( + private boolean handleExistingVirtualComputeGroup( Cloud.ClusterPB clusterInMs, CloudComputeGroupMeta virtualGroupInFe) { if (!isClusterIdConsistent(clusterInMs, virtualGroupInFe)) { - return; + return false; } if (!isClusterPolicyValid(clusterInMs)) { - return; + return false; } if (!areSubComputeGroupsValid(clusterInMs, virtualGroupInFe)) { - return; + return false; } diffAndUpdateComputeGroup(clusterInMs, virtualGroupInFe); + return true; } private boolean isClusterIdConsistent(Cloud.ClusterPB cluster, CloudComputeGroupMeta computeGroup) { @@ -478,27 +508,27 @@ private void diffAndUpdateComputeGroup(Cloud.ClusterPB cluster, CloudComputeGrou } } - private void handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List computeClusters) { + private boolean handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List computeClusters) { List subComputeGroups = cluster.getClusterNamesList(); if (subComputeGroups.isEmpty()) { LOG.info("found virtual cluster {} which has no sub clusters, skip empty virtual cluster", cluster); - return; + return false; } if (subComputeGroups.size() != 2) { LOG.warn("virtual compute err, sub compute group size not eq 2, in ms {}", subComputeGroups); - return; + return false; } if (!cluster.hasClusterPolicy()) { LOG.warn("virtual compute err, no cluster policy {}", cluster); - return; + return false; } if (!cluster.getClusterPolicy().hasActiveClusterName()) { LOG.warn("virtual compute err, active cluster empty in ms {}", cluster); - return; + return false; } if (cluster.getClusterPolicy().getStandbyClusterNamesList().size() != 1) { LOG.warn("virtual compute err, standby cluster size not eq 1 in ms {}", cluster); - return; + return false; } checkSubClusters(subComputeGroups, cluster, computeClusters); CloudComputeGroupMeta computeGroup = new CloudComputeGroupMeta(cluster.getClusterId(), @@ -513,6 +543,7 @@ private void handleNewVirtualComputeGroup(Cloud.ClusterPB cluster, List subClusterNames, Cloud.ClusterPB cluster, diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java index b7ae1308ed5227..e06d413c908f29 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java @@ -628,7 +628,7 @@ public void clearClusterToBe(String cluster) { * * Such an entry is already dead weight: getBackendIdImpl() resolves the backend id, gets null and * falls back to hashReplicaToBe(), so removing it does not change routing. But nothing ever removes - * it either -- dropCluster() only touches CloudSystemInfoService, and the rebalancer only walks the + * it either -- removeComputeGroup() only touches CloudSystemInfoService, and the rebalancer only walks the * compute groups that currently exist -- so entries of dropped compute groups pile up forever, both * in FE heap and in the image (the `bes`/`be` field). * diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java index b99e9967800be8..b0f334d354dc31 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/system/CloudSystemInfoService.java @@ -66,6 +66,7 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.HashSet; @@ -247,17 +248,46 @@ public void addVirtualClusterInfoToMapsNoLock(String clusterId, String clusterNa clusterIdToBackend.computeIfAbsent(clusterId, k -> new ArrayList<>()); } - public void removeVirtualClusterInfoFromMapsNoLock(String clusterId, String clusterName) { - LOG.info("remove virtual cluster info from maps, clusterId={}, clusterName={}", clusterId, clusterName); - clusterIdToBackend.remove(clusterId); - clusterNameToId.remove(clusterName); - } - public void renameVirtualClusterInfoFromMapsNoLock(String clusterId, String oldClusterName, String newClusterName) { LOG.info("remove virtual cluster info from maps, clusterId={}, name from {} to {}", clusterId, oldClusterName, newClusterName); clusterNameToId.put(newClusterName, clusterId); - clusterNameToId.remove(oldClusterName); + clusterNameToId.remove(oldClusterName, clusterId); + } + + // Physical and virtual group checkers fetch and apply snapshots independently. + // For example: + // 1. A current physical snapshot installs name -> newId. + // 2. A delayed virtual snapshot overwrites it with name -> oldId. + // 3. A later virtual snapshot removes oldId and its name mapping. + // The new physical group's metadata and BEs still exist, but its name mapping + // is missing. Normal add/rename detection cannot repair it because neither + // the group ID nor the BE names have changed. + // + // Reconcile name mappings for locally installed groups on every sync cycle. + // Callers must exclude records whose metadata reconciliation was rejected. + // This does not reject stale snapshots; it restores the current mapping once + // a later cycle applies current metadata from the meta service. + public void refreshComputeGroupNames(Collection remoteComputeGroups) { + wlock.lock(); + try { + for (Cloud.ClusterPB group : remoteComputeGroups) { + String id = group.getClusterId(); + // Empty physical groups and rejected virtual groups may not be installed locally. + // Do not create a name pointing to missing metadata. + if (!computeGroupIdToComputeGroup.containsKey(id)) { + continue; + } + String name = group.getClusterName(); + String previousId = clusterNameToId.put(name, id); + if (!id.equals(previousId)) { + LOG.warn("repair compute group name mapping from meta service, name={}, oldId={}, currentId={}", + name, previousId, id); + } + } + } finally { + wlock.unlock(); + } } public CloudComputeGroupMeta getComputeGroupByName(String computeGroupName) { @@ -421,11 +451,17 @@ public String ownedByVirtualComputeGroup(String computeGroupName) { } } + // Remove local metadata for a physical or virtual compute group. Physical backends + // must be removed by the caller before removing the group. public void removeComputeGroup(String computeGroupId, String computeGroupName) { try { wlock.lock(); + LOG.info("remove compute group, id={}, name={}", computeGroupId, computeGroupName); computeGroupIdToComputeGroup.remove(computeGroupId); - removeVirtualClusterInfoFromMapsNoLock(computeGroupId, computeGroupName); + clusterIdToBackend.remove(computeGroupId); + // Earlier checkers could publish aliases from rejected renames. Remove every name + // still owned by this ID, without removing names already reused by another group. + clusterNameToId.entrySet().removeIf(entry -> computeGroupId.equals(entry.getValue())); invalidateCloudColocatePlacement(computeGroupId); } finally { wlock.unlock(); @@ -632,22 +668,13 @@ public void updateCloudClusterMapNoLock(List toAdd, List toDel be = be.stream().filter(i -> !d.contains(i.getId())).collect(Collectors.toList()); // ATTN: clusterId may have zero nodes clusterIdToBackend.replace(clusterId, be); - // such as dropCluster, but no lock // ATTN: Empty clusters are treated as dropped clusters. if (be.isEmpty()) { LOG.info("del clusterId {} and clusterName {} due to be nodes eq 0", clusterId, clusterName); MetricRepo.unregisterCloudMetrics(clusterId, clusterName, toDel); - boolean succ = clusterNameToId.remove(clusterName, clusterId); // remove from computeGroupIdToComputeGroup removeComputeGroup(clusterId, clusterName); - - if (!succ) { - LOG.warn("impossible, somewhere err, clusterNameToId {}, " - + "want remove cluster name {}, cluster id {}", - clusterNameToId, clusterName, clusterId); - } - clusterIdToBackend.remove(clusterId); } LOG.info("update (del) cloud cluster map, clusterName={} clusterId={} backendNum={} current backend={}", clusterName, clusterId, be.size(), b); @@ -1220,7 +1247,7 @@ public void updateClusterNameToId(final String newName, final String originalName, final String clusterId) { wlock.lock(); try { - clusterNameToId.remove(originalName); + clusterNameToId.remove(originalName, clusterId); clusterNameToId.put(newName, clusterId); } finally { wlock.unlock(); @@ -1252,17 +1279,6 @@ public String getClusterNameByClusterIdNoLock(final String clusterId) { return clusterName; } - public void dropCluster(final String clusterId, final String clusterName) { - wlock.lock(); - try { - clusterNameToId.remove(clusterName, clusterId); - clusterIdToBackend.remove(clusterId); - invalidateCloudColocatePlacement(clusterId); - } finally { - wlock.unlock(); - } - } - public List getCloudClusterNames() { rlock.lock(); try { diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudClusterCheckerTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudClusterCheckerTest.java new file mode 100644 index 00000000000000..bd20866e1598f1 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudClusterCheckerTest.java @@ -0,0 +1,308 @@ +// 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.doris.cloud.catalog; + +import org.apache.doris.catalog.Env; +import org.apache.doris.cloud.CacheHotspotManager; +import org.apache.doris.cloud.proto.Cloud; +import org.apache.doris.cloud.system.CloudSystemInfoService; +import org.apache.doris.common.Config; +import org.apache.doris.persist.EditLog; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.lang.reflect.Method; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +class CloudClusterCheckerTest { + private String originalCloudUniqueId; + private CloudSystemInfoService service; + private CloudEnv env; + private MockedStatic mockedEnv; + + @BeforeEach + void setUp() { + originalCloudUniqueId = Config.cloud_unique_id; + Config.cloud_unique_id = "test_cloud"; + service = Mockito.spy(new CloudSystemInfoService()); + env = Mockito.mock(CloudEnv.class); + AtomicLong nextId = new AtomicLong(10000); + Mockito.when(env.getNextId()).thenAnswer(invocation -> nextId.incrementAndGet()); + Mockito.when(env.getEditLog()).thenReturn(Mockito.mock(EditLog.class)); + Mockito.when(env.getCacheHotspotMgr()).thenReturn(Mockito.mock(CacheHotspotManager.class)); + mockedEnv = mockEnv(); + } + + private MockedStatic mockEnv() { + MockedStatic mock = Mockito.mockStatic(Env.class); + mock.when(Env::getCurrentEnv).thenReturn(env); + mock.when(Env::getCurrentSystemInfo).thenReturn(service); + return mock; + } + + @AfterEach + void tearDown() { + mockedEnv.close(); + Config.cloud_unique_id = originalCloudUniqueId; + } + + @Test + void testCheckerDeletesRecreatedGroupById() throws Exception { + Cloud.ClusterPB oldGroup = physical("old", "reused", "127.0.0.1"); + Cloud.ClusterPB newGroup = physical("new", "reused", "127.0.0.2"); + syncPhysical(oldGroup); + long oldBackendId = service.getCloudClusterIdToBackend(false).get("old").get(0).getId(); + + syncPhysical(newGroup); + + Assertions.assertNull(service.getBackend(oldBackendId)); + Assertions.assertNull(service.getComputeGroupById("old")); + Assertions.assertFalse(service.getCloudClusterIdToBackend(false).containsKey("old")); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + syncPhysical(newGroup); + Assertions.assertEquals(1, service.getAllBackendsByAllCluster().size()); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + } + + @Test + void testCheckerRenamesGroupAfterOldNameIsReused() throws Exception { + Cloud.ClusterPB oldGroup = physical("old", "reused", "127.0.0.1"); + Cloud.ClusterPB renamed = oldGroup.toBuilder().setClusterName("renamed").build(); + Cloud.ClusterPB replacement = physical("new", "reused", "127.0.0.2"); + syncPhysical(oldGroup); + syncPhysical(renamed, replacement); + + Assertions.assertEquals("old", service.getCloudClusterIdByName("renamed")); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + syncPhysical(renamed, replacement); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + } + + @Test + void testPhysicalRenameDoesNotRemoveAnotherGroupsName() { + service.addComputeGroup("old", new CloudComputeGroupMeta("old", "reused", + CloudComputeGroupMeta.ComputeTypeEnum.COMPUTE)); + service.addComputeGroup("new", new CloudComputeGroupMeta("new", "reused", + CloudComputeGroupMeta.ComputeTypeEnum.COMPUTE)); + service.updateClusterNameToId("renamed", "reused", "old"); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + Assertions.assertEquals("old", service.getCloudClusterIdByName("renamed")); + } + + @Test + void testVirtualRenameDoesNotRemoveAnotherGroupsName() { + service.addComputeGroup("old", new CloudComputeGroupMeta("old", "reused", + CloudComputeGroupMeta.ComputeTypeEnum.VIRTUAL)); + service.addComputeGroup("new", new CloudComputeGroupMeta("new", "reused", + CloudComputeGroupMeta.ComputeTypeEnum.VIRTUAL)); + service.renameVirtualComputeGroup("old", "reused", new CloudComputeGroupMeta("old", "renamed", + CloudComputeGroupMeta.ComputeTypeEnum.VIRTUAL)); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + Assertions.assertEquals("old", service.getCloudClusterIdByName("renamed")); + } + + @Test + void testLaterCycleRepairsNameAfterStaleVirtualSnapshot() throws Exception { + Cloud.ClusterPB active = physical("active", "active", "127.0.0.1"); + Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2"); + Cloud.ClusterPB replacement = physical("new", "reused", "127.0.0.3"); + syncPhysical(active, standby); + Cloud.GetInstanceResponse stale = instance(active, standby, virtual("old", "reused")); + CountDownLatch responseFetched = new CountDownLatch(1); + CountDownLatch applyResponse = new CountDownLatch(1); + Mockito.doAnswer(invocation -> { + responseFetched.countDown(); + Assertions.assertTrue(applyResponse.await(30, TimeUnit.SECONDS)); + return stale; + }).when(service).getCloudInstance(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future staleCycle = executor.submit(() -> { + try (MockedStatic ignored = mockEnv()) { + new CloudInstanceStatusChecker(service).runAfterCatalogReady(); + } + }); + Assertions.assertTrue(responseFetched.await(30, TimeUnit.SECONDS)); + syncPhysical(active, standby, replacement); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + applyResponse.countDown(); + staleCycle.get(30, TimeUnit.SECONDS); + Assertions.assertEquals("old", service.getCloudClusterIdByName("reused")); + + // A current physical snapshot must repair even a non-empty, wrong mapping. + syncPhysical(active, standby, replacement); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + syncInstance(active, standby, replacement); + Assertions.assertNull(service.getComputeGroupById("old")); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + syncPhysical(active, standby, replacement); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + } finally { + applyResponse.countDown(); + executor.shutdownNow(); + Assertions.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + + @Test + void testLaterCyclesRepairMissingPhysicalAndVirtualNames() throws Exception { + Cloud.ClusterPB active = physical("active", "active", "127.0.0.1"); + Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2"); + Cloud.ClusterPB current = physical("new", "reused", "127.0.0.3"); + syncPhysical(active, standby, current); + syncInstance(active, standby, virtual("old", "reused")); + syncInstance(active, standby, current); + Assertions.assertNull(service.getCloudClusterIdByName("reused")); + syncPhysical(active, standby, current); + Assertions.assertEquals("new", service.getCloudClusterIdByName("reused")); + + // Reverse the types: an obsolete physical snapshot overwrites a current virtual group. + syncPhysical(active, standby); + Cloud.ClusterPB currentVirtual = virtual("virtual", "reused"); + syncInstance(active, standby, currentVirtual); + syncPhysical(active, standby, current); + syncPhysical(active, standby); + Assertions.assertNull(service.getCloudClusterIdByName("reused")); + syncInstance(active, standby, currentVirtual); + Assertions.assertEquals("virtual", service.getCloudClusterIdByName("reused")); + syncInstance(active, standby, currentVirtual); + Assertions.assertEquals("virtual", service.getCloudClusterIdByName("reused")); + Assertions.assertNull(service.getComputeGroupById("new")); + } + + @Test + void testRejectedVirtualPolicyDoesNotPublishRenamedGroup() throws Exception { + assertRejectedVirtualRename(virtual("virtual", "renamed").toBuilder() + .clearClusterPolicy().build()); + } + + @Test + void testRejectedVirtualSubgroupsDoNotPublishRenamedGroup() throws Exception { + assertRejectedVirtualRename(virtual("virtual", "renamed").toBuilder() + .clearClusterNames().addClusterNames("active").build()); + } + + private void assertRejectedVirtualRename(Cloud.ClusterPB rejected) throws Exception { + Cloud.ClusterPB active = physical("active", "active", "127.0.0.1"); + Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2"); + syncPhysical(active, standby); + syncInstance(active, standby, virtual("virtual", "original")); + + syncInstance(active, standby, rejected); + + Assertions.assertEquals("original", service.getComputeGroupById("virtual").getName()); + Assertions.assertEquals("virtual", service.getCloudClusterIdByName("original")); + Assertions.assertNull(service.getCloudClusterIdByName("renamed")); + + // The rejected record must not make the still-present group look obsolete. + // Once it actually disappears from MS, neither name may point to its removed ID. + syncInstance(active, standby); + Assertions.assertNull(service.getComputeGroupById("virtual")); + Assertions.assertNull(service.getCloudClusterIdByName("original")); + Assertions.assertNull(service.getCloudClusterIdByName("renamed")); + } + + @Test + void testAcceptedVirtualRenamePublishesNewName() throws Exception { + Cloud.ClusterPB active = physical("active", "active", "127.0.0.1"); + Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2"); + syncPhysical(active, standby); + syncInstance(active, standby, virtual("virtual", "original")); + + syncInstance(active, standby, virtual("virtual", "renamed")); + + Assertions.assertEquals("renamed", service.getComputeGroupById("virtual").getName()); + Assertions.assertEquals("virtual", service.getCloudClusterIdByName("renamed")); + Assertions.assertNull(service.getCloudClusterIdByName("original")); + } + + @Test + void testRejectedNewVirtualGroupDoesNotPublishName() { + syncInstance(virtual("virtual", "rejected").toBuilder().clearClusterPolicy().build()); + + Assertions.assertNull(service.getComputeGroupById("virtual")); + Assertions.assertNull(service.getCloudClusterIdByName("rejected")); + } + + @Test + void testVirtualRemovalClearsStaleAliasesButKeepsReusedName() throws Exception { + Cloud.ClusterPB active = physical("active", "active", "127.0.0.1"); + Cloud.ClusterPB standby = physical("standby", "standby", "127.0.0.2"); + syncPhysical(active, standby); + syncInstance(active, standby, virtual("virtual", "reused")); + // Simulate an alias left by an earlier checker that refreshed a rejected rename. + service.addVirtualClusterInfoToMapsNoLock("virtual", "stale_alias"); + syncPhysical(active, standby, physical("replacement", "reused", "127.0.0.3")); + + syncInstance(active, standby); + + Assertions.assertNull(service.getComputeGroupById("virtual")); + Assertions.assertNull(service.getCloudClusterIdByName("stale_alias")); + Assertions.assertEquals("replacement", service.getCloudClusterIdByName("reused")); + } + + private void syncPhysical(Cloud.ClusterPB... groups) throws Exception { + Mockito.doReturn(Cloud.GetClusterResponse.newBuilder().setStatus(ok()) + .addAllCluster(List.of(groups)).build()).when(service).getCloudCluster("", "", ""); + // The production reconciliation entry, including add, delete, diff and validation. + Method method = CloudClusterChecker.class.getDeclaredMethod("checkCloudBackends"); + method.setAccessible(true); + method.invoke(new CloudClusterChecker(service)); + } + + private void syncInstance(Cloud.ClusterPB... groups) { + Mockito.doReturn(instance(groups)).when(service).getCloudInstance(); + new CloudInstanceStatusChecker(service).runAfterCatalogReady(); + } + + private Cloud.GetInstanceResponse instance(Cloud.ClusterPB... groups) { + return Cloud.GetInstanceResponse.newBuilder().setStatus(ok()) + .setInstance(Cloud.InstanceInfoPB.newBuilder().setStatus(Cloud.InstanceInfoPB.Status.NORMAL) + .addAllClusters(List.of(groups))).build(); + } + + private Cloud.MetaServiceResponseStatus ok() { + return Cloud.MetaServiceResponseStatus.newBuilder().setCode(Cloud.MetaServiceCode.OK).build(); + } + + private Cloud.ClusterPB physical(String id, String name, String ip) { + return Cloud.ClusterPB.newBuilder().setClusterId(id).setClusterName(name) + .setType(Cloud.ClusterPB.Type.COMPUTE).setClusterStatus(Cloud.ClusterStatus.NORMAL) + .addNodes(Cloud.NodeInfoPB.newBuilder().setIp(ip).setHeartbeatPort(9050) + .setCloudUniqueId(id).setStatus(Cloud.NodeStatusPB.NODE_STATUS_RUNNING)).build(); + } + + private Cloud.ClusterPB virtual(String id, String name) { + return Cloud.ClusterPB.newBuilder().setClusterId(id).setClusterName(name) + .setType(Cloud.ClusterPB.Type.VIRTUAL).addAllClusterNames(List.of("active", "standby")) + .setClusterPolicy(Cloud.ClusterPolicy.newBuilder().setType(Cloud.ClusterPolicy.PolicyType.ActiveStandby) + .setActiveClusterName("active") + .addStandbyClusterNames("standby")).build(); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/system/CloudSystemInfoServiceTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/system/CloudSystemInfoServiceTest.java index 18734d8a1b0705..5962833c4c4da3 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/cloud/system/CloudSystemInfoServiceTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/system/CloudSystemInfoServiceTest.java @@ -1062,6 +1062,49 @@ public void testContainsCloudCluster() { Assertions.assertFalse(infoService.containsCloudCluster("cluster_2")); } + @Test + public void testRemovingRecreatedComputeGroupKeepsCurrentNameMapping() { + infoService = new CloudSystemInfoService(); + String clusterName = "recreated_cluster"; + String oldClusterId = "old_cluster_id"; + String newClusterId = "new_cluster_id"; + + Backend oldBackend = new Backend(1L, "127.0.0.1", 9050); + Map oldTagMap = Tag.DEFAULT_BACKEND_TAG.toMap(); + oldTagMap.put(Tag.CLOUD_CLUSTER_NAME, clusterName); + oldTagMap.put(Tag.CLOUD_CLUSTER_ID, oldClusterId); + oldBackend.setTagMap(oldTagMap); + infoService.updateCloudClusterMapNoLock(List.of(oldBackend), new ArrayList<>()); + + Backend newBackend = new Backend(2L, "127.0.0.2", 9050); + Map newTagMap = Tag.DEFAULT_BACKEND_TAG.toMap(); + newTagMap.put(Tag.CLOUD_CLUSTER_NAME, clusterName); + newTagMap.put(Tag.CLOUD_CLUSTER_ID, newClusterId); + newBackend.setTagMap(newTagMap); + infoService.updateCloudClusterMapNoLock(List.of(newBackend), new ArrayList<>()); + + infoService.updateCloudClusterMapNoLock(new ArrayList<>(), List.of(oldBackend)); + + Assertions.assertEquals(newClusterId, infoService.getCloudClusterIdByName(clusterName)); + Assertions.assertNull(infoService.getComputeGroupById(oldClusterId)); + Assertions.assertNotNull(infoService.getComputeGroupById(newClusterId)); + } + + @Test + public void testRemovingCurrentComputeGroupRemovesNameMapping() { + infoService = new CloudSystemInfoService(); + String clusterName = "removed_cluster"; + String clusterId = "removed_cluster_id"; + CloudComputeGroupMeta computeGroup = new CloudComputeGroupMeta( + clusterId, clusterName, CloudComputeGroupMeta.ComputeTypeEnum.COMPUTE); + infoService.addComputeGroup(clusterId, computeGroup); + + infoService.removeComputeGroup(clusterId, clusterName); + + Assertions.assertNull(infoService.getCloudClusterIdByName(clusterName)); + Assertions.assertNull(infoService.getComputeGroupById(clusterId)); + } + /** * Helper method to create a test ConnectContext with specific cluster name */ diff --git a/regression-test/data/cloud_p0/node_mgr/test_compute_group_name_reconciliation.out b/regression-test/data/cloud_p0/node_mgr/test_compute_group_name_reconciliation.out new file mode 100644 index 00000000000000..889a512b525aa4 --- /dev/null +++ b/regression-test/data/cloud_p0/node_mgr/test_compute_group_name_reconciliation.out @@ -0,0 +1,21 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !recreated -- +1 +2 +3 + +-- !renamed -- +1 +2 +3 + +-- !reused -- +1 +2 +3 + +-- !repaired -- +1 +2 +3 + diff --git a/regression-test/suites/cloud_p0/node_mgr/test_compute_group_name_reconciliation.groovy b/regression-test/suites/cloud_p0/node_mgr/test_compute_group_name_reconciliation.groovy new file mode 100644 index 00000000000000..ac1e7b586f3d48 --- /dev/null +++ b/regression-test/suites/cloud_p0/node_mgr/test_compute_group_name_reconciliation.groovy @@ -0,0 +1,169 @@ +// 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. + +import groovy.json.JsonOutput +import org.apache.doris.regression.suite.ClusterOptions + +// Run with a cloud image containing the checker debug points below. For a pre-fix +// reproduction, keep the debug points but revert the reconciliation changes. +suite('test_compute_group_name_reconciliation', 'cloud_p0,docker') { + if (!isCloudMode()) { + return + } + + def options = new ClusterOptions() + options.cloudMode = true + options.setFeNum(1) + options.setBeNum(1) + options.enableDebugPoints() + options.feConfigs += ['cloud_cluster_check_interval_second=1', 'heartbeat_interval_second=1'] + + docker(options) { + def ms = cluster.getAllMetaservices().get(0) + def msEndpoint = "${ms.host}:${ms.httpPort}" + def feLog = new File(cluster.getFeByIndex(1).getLogFilePath()) + def physicalPause = 'CloudClusterChecker.checkCloudBackends.pause' + def instancePause = 'CloudInstanceStatusChecker.runAfterCatalogReady.pause' + def stalePause = 'CloudInstanceStatusChecker.afterGetInstance.pause' + def debug = GetDebugPoint() + + def alterCluster = { operation, group -> + httpTest { + endpoint msEndpoint + uri "/MetaService/http/${operation}?token=${token}" + body JsonOutput.toJson([instance_id: 'default_instance_id', cluster: group]) + check { code, response -> + def result = parseJson(response) + assertTrue(code == 200 && result.code == 'OK', "${operation}: ${response}") + } + } + } + def remoteGroup = { name -> + def group = get_instance(ms).clusters.find { it.cluster_name == name } + assertNotNull(group, "remote compute group ${name}") + group + } + def pause = { point, phase, params = [:] -> + debug.enableDebugPointForAllFEs(point, params + [phase: phase, timeout: '180']) + // A unique phase proves this cycle reached the pause; no timing guesses. + awaitUntil(30) { feLog.text.contains("${point} phase=${phase}") } + } + def resume = { point -> debug.disableDebugPointForAllFEs(point) } + def waitPhysical = { name, id, removedIds -> + awaitUntil(60) { + def bes = sql_return_maparray('SHOW BACKENDS') + def tags = bes.collect { parseJson(it.Tag) } + def groups = sql_return_maparray('SHOW COMPUTE GROUPS') + tags.any { it.compute_group_id == id && it.compute_group_name == name } && + !tags.any { removedIds.contains(it.compute_group_id) } && + groups.any { it.Name == name && it.BackendNum.toInteger() == 1 } + } + sql "USE @${name}" + } + + try { + cluster.addBackend(1, 'cg_reused') + cluster.addBackend(1, 'cg_spare') + def old = remoteGroup('cg_reused') + def spare = remoteGroup('cg_spare') + waitPhysical('cg_reused', old.cluster_id, []) + waitPhysical('cg_spare', spare.cluster_id, []) + sql 'USE @compute_cluster' + sql 'DROP TABLE IF EXISTS test_compute_group_name_reconciliation' + sql '''CREATE TABLE test_compute_group_name_reconciliation (k INT) + DISTRIBUTED BY HASH(k) BUCKETS 1 PROPERTIES ('replication_num'='1')''' + sql 'INSERT INTO test_compute_group_name_reconciliation VALUES (1), (2), (3)' + + // 1. A single checker cycle sees a same-name replacement on distinct BEs. + pause(physicalPause, 'recreate') + alterCluster('drop_cluster', [cluster_id: old.cluster_id, cluster_name: 'cg_reused']) + alterCluster('drop_cluster', [cluster_id: spare.cluster_id, cluster_name: 'cg_spare']) + alterCluster('add_cluster', [cluster_id: 'cg_recreated_id', cluster_name: 'cg_reused', + type: 'COMPUTE', nodes: spare.nodes]) + resume(physicalPause) + waitPhysical('cg_reused', 'cg_recreated_id', [old.cluster_id, spare.cluster_id]) + order_qt_recreated 'SELECT k FROM test_compute_group_name_reconciliation' + + // 2. Rename A and reuse its old name for B before the next checker cycle. + pause(physicalPause, 'rename') + alterCluster('rename_cluster', [cluster_id: 'cg_recreated_id', cluster_name: 'cg_renamed']) + alterCluster('add_cluster', [cluster_id: 'cg_reuse_after_rename_id', cluster_name: 'cg_reused', + type: 'COMPUTE', nodes: old.nodes]) + resume(physicalPause) + waitPhysical('cg_renamed', 'cg_recreated_id', [old.cluster_id, spare.cluster_id]) + order_qt_renamed 'SELECT k FROM test_compute_group_name_reconciliation' + waitPhysical('cg_reused', 'cg_reuse_after_rename_id', [old.cluster_id, spare.cluster_id]) + order_qt_reused 'SELECT k FROM test_compute_group_name_reconciliation' + + // 3. Hold an old virtual-group response while the physical checker installs + // a newer same-name physical group. Then let the old response overwrite it. + cluster.addBackend(1, 'cg_snapshot_spare') + def snapshotSpare = remoteGroup('cg_snapshot_spare') + waitPhysical('cg_snapshot_spare', snapshotSpare.cluster_id, []) + sql 'USE @compute_cluster' + pause(instancePause, 'before_virtual') + alterCluster('add_cluster', [cluster_id: 'cg_stale_virtual_id', cluster_name: 'cg_snapshot', + type: 'VIRTUAL', cluster_names: ['compute_cluster', 'cg_renamed'], + cluster_policy: [type: 'ActiveStandby', active_cluster_name: 'compute_cluster', + standby_cluster_names: ['cg_renamed']]]) + debug.enableDebugPointForAllFEs(stalePause, + [cluster_id: 'cg_stale_virtual_id', phase: 'stale_response', timeout: '180']) + resume(instancePause) + awaitUntil(30) { feLog.text.contains("${stalePause} phase=stale_response") } + alterCluster('drop_cluster', [cluster_id: 'cg_stale_virtual_id', cluster_name: 'cg_snapshot']) + alterCluster('drop_cluster', [cluster_id: snapshotSpare.cluster_id, cluster_name: 'cg_snapshot_spare']) + alterCluster('add_cluster', [cluster_id: 'cg_current_physical_id', cluster_name: 'cg_snapshot', + type: 'COMPUTE', nodes: snapshotSpare.nodes]) + waitPhysical('cg_snapshot', 'cg_current_physical_id', [snapshotSpare.cluster_id]) + sql 'USE @compute_cluster' + pause(physicalPause, 'hold_repair') + debug.enableDebugPointForAllFEs(instancePause, [phase: 'after_stale', timeout: '180']) + resume(stalePause) + awaitUntil(30) { feLog.text.contains("${instancePause} phase=after_stale") } + awaitUntil(30) { + sql_return_maparray('SHOW COMPUTE GROUPS').any { + it.Name == 'cg_snapshot' && it.SubComputeGroups.contains('cg_renamed') + } + } + + // Current instance data removes V. P's BEs remain, but its name is now missing. + resume(instancePause) + awaitUntil(60) { + !sql_return_maparray('SHOW COMPUTE GROUPS').any { it.Name == 'cg_snapshot' } + } + // No node or name change in MS: only periodic mapping repair can restore P. + resume(physicalPause) + waitPhysical('cg_snapshot', 'cg_current_physical_id', [snapshotSpare.cluster_id]) + order_qt_repaired 'SELECT k FROM test_compute_group_name_reconciliation' + + // Check stability after several subsequent physical AND instance cycles. + 3.times { + def logOffset = feLog.text.length() + awaitUntil(30) { + def subsequent = feLog.text.substring(logOffset) + subsequent.contains('daemon cluster get cluster info succ') && + subsequent.contains('finished to cloud instance checker') + } + waitPhysical('cg_snapshot', 'cg_current_physical_id', [snapshotSpare.cluster_id]) + } + } finally { + resume(stalePause) + resume(instancePause) + resume(physicalPause) + } + } +}