Merge branch 'fix-ZSTAC-81028' into '5.5.0'

<fix>[migration]: fix batch migration error

See merge request zstackio/zstack!8986

Author: gitlab

plugin/zbs/src/main/java/org/zstack/storage/zbs/ZbsStorageController.java

@@ -4,7 +4,6 @@
 import org.springframework.beans.factory.annotation.Autowire;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Configurable;
-import org.zstack.cbd.*;
 import org.zstack.cbd.kvm.CbdHeartbeatVolumeTO;
 import org.zstack.cbd.kvm.CbdVolumeTo;
 import org.zstack.compute.host.HostGlobalConfig;
@@ -32,7 +31,6 @@
 import org.zstack.header.host.HostVO;
 import org.zstack.header.image.ImageConstant;
 import org.zstack.header.message.MessageReply;
-import org.zstack.header.rest.RESTFacade;
 import org.zstack.header.storage.addon.*;
 import org.zstack.header.storage.addon.primary.*;
 import org.zstack.header.storage.primary.*;
@@ -711,9 +709,8 @@ private LogicalPoolInfo allocateFreePool(long size, Predicate<LogicalPoolInfo> f
     }
 
     private List<LogicalPoolInfo> getSelfPools() {
-        List<LogicalPoolInfo> logicalPoolInfos = addonInfo.getLogicalPoolInfos();
-        logicalPoolInfos.removeIf(it -> !config.getPoolNames().contains(it.getLogicalPoolName()));
-        return logicalPoolInfos;
+        return addonInfo.getLogicalPoolInfos().stream().filter(pool ->
+                config.getPoolNames().contains(pool.getLogicalPoolName())).collect(Collectors.toList());
     }
 
     @Override

storage/src/main/java/org/zstack/storage/primary/PrimaryStorageBase.java

@@ -2,6 +2,7 @@
 
 import com.google.common.collect.Interner;
 import com.google.common.collect.Interners;
+import java.util.function.Consumer;
 import org.springframework.beans.factory.annotation.Autowire;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Configurable;
@@ -19,6 +20,9 @@
 import org.zstack.core.job.JobQueueFacade;
 import org.zstack.core.thread.ChainTask;
 import org.zstack.core.thread.MergeQueue;
+import org.zstack.core.thread.SingleFlightTask;
+import org.zstack.core.thread.SingleFlightTask.SingleFlightDone;
+import org.zstack.core.thread.SingleFlightTaskResult;
 import org.zstack.core.thread.SyncTaskChain;
 import org.zstack.core.thread.ThreadFacade;
 import org.zstack.core.trash.StorageTrash;
@@ -1244,23 +1248,33 @@ public void setup() {
 
                     @Override
                     public void run(final FlowTrigger trigger, Map data) {
-                        syncPhysicalCapacity(new ReturnValueCompletion<PhysicalCapacityUsage>(trigger) {
-                            @Override
-                            public void success(PhysicalCapacityUsage returnValue) {
-                                PrimaryStorageCapacityUpdater updater = new PrimaryStorageCapacityUpdater(self.getUuid());
-                                updater.run(cap -> {
-                                    cap.setAvailablePhysicalCapacity(returnValue.availablePhysicalSize < 0 ? 0 : returnValue.availablePhysicalSize);
-                                    cap.setTotalPhysicalCapacity(returnValue.totalPhysicalSize);
-                                    return cap;
-                                });
-                                trigger.next();
-                            }
-
-                            @Override
-                            public void fail(ErrorCode errorCode) {
-                                trigger.fail(errorCode);
-                            }
-                        });
+                        thdf.singleFlightSubmit(new SingleFlightTask(trigger)
+                                .setSyncSignature("sync-physical-capacity-on-ps-" + self.getUuid())
+                                .run(completion -> {
+                                    syncPhysicalCapacity(new ReturnValueCompletion<PhysicalCapacityUsage>(completion) {
+                                        @Override
+                                        public void success(PhysicalCapacityUsage returnValue) {
+                                            PrimaryStorageCapacityUpdater updater = new PrimaryStorageCapacityUpdater(self.getUuid());
+                                            updater.run(cap -> {
+                                                cap.setAvailablePhysicalCapacity(returnValue.availablePhysicalSize < 0 ? 0 : returnValue.availablePhysicalSize);
+                                                cap.setTotalPhysicalCapacity(returnValue.totalPhysicalSize);
+                                                return cap;
+                                            });
+                                            completion.success(null);
+                                        }
+
+                                        @Override
+                                        public void fail(ErrorCode errorCode) {
+                                            completion.fail(errorCode);
+                                        }
+                                    });
+                                }).done(result -> {
+                                    if (!result.isSuccess()) {
+                                        trigger.fail(result.getErrorCode());
+                                    } else {
+                                        trigger.next();
+                                    }
+                                }));
                     }
                 });
 

test/src/test/groovy/org/zstack/test/integration/storage/primary/addon/zbs/ZbsPrimaryStorageCase.groovy

@@ -32,6 +32,8 @@ import org.zstack.testlib.SubCase
 import org.zstack.utils.data.SizeUnit
 import org.zstack.utils.gson.JSONObjectUtil
 
+import java.util.concurrent.atomic.AtomicInteger
+
 /**
  * @author Xingwei Yu
  * @date 2024/4/19 10:09
@@ -164,6 +166,7 @@ class ZbsPrimaryStorageCase extends SubCase {
             kvm = env.inventoryByName("kvm-1") as KVMHostInventory
             evtf = bean(EventFacade.class)
 
+            testSyncPrimaryStorageCapacityConcurrently()
             testDefaultConfig()
             testUpdateExternalPrimaryStorage()
             testMdsConnectFailed()
@@ -177,6 +180,22 @@ class ZbsPrimaryStorageCase extends SubCase {
         }
     }
 
+    void testSyncPrimaryStorageCapacityConcurrently() {
+        def threads = new ArrayList<>()
+        def success_cnt = new AtomicInteger(0)
+        (1..20).forEach {
+            threads.add(Thread.start {
+                syncPrimaryStorageCapacity {
+                    primaryStorageUuid = ps.uuid
+                }
+                success_cnt.incrementAndGet()
+            })
+        }
+
+        threads.each { it.join() }
+        assert success_cnt.get() == 20
+    }
+
     void testCheckHostStorageConnection() {
         attachPrimaryStorageToCluster {
             primaryStorageUuid = ps.uuid