fix(portForwarding): serialize concurrent PF rule creation per VIP to prevent duplicates

compute/src/main/java/org/zstack/compute/vm/VmNicManagerImpl.java

@@ -58,6 +58,10 @@ public void afterAddIpAddress(String vmNicUUid, String usedIpUuid) {
         SQL.New(UsedIpVO.class).eq(UsedIpVO_.uuid, usedIpUuid).set(UsedIpVO_.vmNicUuid, vmNicUUid).update();
 
         VmNicVO nic = Q.New(VmNicVO.class).eq(VmNicVO_.uuid, vmNicUUid).find();
+        if (nic == null) {
+            logger.debug(String.format("VmNic[uuid:%s] not found, skip afterAddIpAddress", vmNicUUid));
+            return;
+        }
 
         UsedIpVO temp = null;
         /* if there is ipv4 addresses, we put the first attached ipv4 address to VmNic.ip
@@ -88,6 +92,10 @@ public void afterAddIpAddress(String vmNicUUid, String usedIpUuid) {
     @Override
     public void afterDelIpAddress(String vmNicUUid, String usedIpUuid) {
         VmNicVO nic = Q.New(VmNicVO.class).eq(VmNicVO_.uuid, vmNicUUid).find();
+        if (nic == null) {
+            logger.debug(String.format("VmNic[uuid:%s] not found, skip afterDelIpAddress", vmNicUUid));
+            return;
+        }
         if (nic.getUsedIpUuid() != null && !nic.getUsedIpUuid().equals(usedIpUuid)) {
             return;
         }

conf/i18n/globalErrorCodeMapping/global-error-en_US.json

@@ -3374,7 +3374,7 @@
   "ORG_ZSTACK_NETWORK_HUAWEI_IMASTER_10019": "delete token of SDN controller [IP:%s] failed because %s",
   "ORG_ZSTACK_STORAGE_PRIMARY_BLOCK_10004": "Cannot execute volume mapping to host flow due to invalid volume ID.%s",
   "ORG_ZSTACK_NETWORK_SERVICE_PORTFORWARDING_10007": "port forwarding rule [uuid:%s] has not been attached to any virtual machine network interface, cannot detach",
-  "ORG_ZSTACK_MEVOCO_10088": "cannot take a snapshot for volumes[%s] when volume[uuid: %s] is not attached",
+  "ORG_ZSTACK_MEVOCO_10088": "cannot create snapshot for volume[uuid:%s] because it is not attached to any VM instance. Please attach the volume to a VM first. Affected volumes: %s",
   "ORG_ZSTACK_STORAGE_PRIMARY_BLOCK_10005": "Cannot execute map LUN to host flow due to invalid LUN type: %s",
   "ORG_ZSTACK_NETWORK_SERVICE_PORTFORWARDING_10008": "port forwarding rule [uuid:%s] has been associated with vm nic [uuid:%s], cannot be reassigned again",
   "ORG_ZSTACK_MEVOCO_10087": "A Running VM[uuid:%s] has no associated Host UUID.",

conf/i18n/globalErrorCodeMapping/global-error-zh_CN.json

@@ -3374,7 +3374,7 @@
   "ORG_ZSTACK_NETWORK_HUAWEI_IMASTER_10019": "删除 SDN 控制器 [IP：%s] 的令牌失败，因为 %s",
   "ORG_ZSTACK_STORAGE_PRIMARY_BLOCK_10004": "无法执行映射LUN到主机流程，无效的LUN ID",
   "ORG_ZSTACK_NETWORK_SERVICE_PORTFORWARDING_10007": "端口转发规则 rule[uuid:%s] 没有绑定到任何 VM 的网卡上，无法解除绑定",
-  "ORG_ZSTACK_MEVOCO_10088": "无法为挂载状态以外的卷[%s]创建快照",
+  "ORG_ZSTACK_MEVOCO_10088": "无法为云盘[uuid:%s]创建快照，因为该云盘未挂载到任何云主机。请先将云盘挂载到云主机后再创建快照。相关云盘: %s",
   "ORG_ZSTACK_STORAGE_PRIMARY_BLOCK_10005": "无法执行映射LUN到主机流程，无效的LUN类型",
   "ORG_ZSTACK_NETWORK_SERVICE_PORTFORWARDING_10008": "端口转发规则[uuid:%s]已绑定到VM网卡[uuid:%s]，无法再次绑定",
   "ORG_ZSTACK_MEVOCO_10087": "如何一个运行中的VM[uuid:%s]没有宿主机uuid?",

core/src/main/java/org/zstack/core/cloudbus/ResourceDestinationMakerImpl.java

@@ -27,27 +27,27 @@ public class ResourceDestinationMakerImpl implements ManagementNodeChangeListene
     private DatabaseFacade dbf;
 
     @Override
-    public void nodeJoin(ManagementNodeInventory inv) {
+    public synchronized void nodeJoin(ManagementNodeInventory inv) {
         nodeHash.add(inv.getUuid());
         nodes.put(inv.getUuid(), new NodeInfo(inv));
     }
 
     @Override
-    public void nodeLeft(ManagementNodeInventory inv) {
+    public synchronized void nodeLeft(ManagementNodeInventory inv) {
         String nodeId = inv.getUuid();
         nodeHash.remove(nodeId);
         nodes.remove(nodeId);
     }
 
     @Override
-    public void iAmDead(ManagementNodeInventory inv) {
+    public synchronized void iAmDead(ManagementNodeInventory inv) {
         String nodeId = inv.getUuid();
         nodeHash.remove(nodeId);
         nodes.remove(nodeId);
     }
 
     @Override
-    public void iJoin(ManagementNodeInventory inv) {
+    public synchronized void iJoin(ManagementNodeInventory inv) {
         List<ManagementNodeVO> lst = Q.New(ManagementNodeVO.class).list();
         lst.forEach((ManagementNodeVO node) -> {
             nodeHash.add(node.getUuid());
@@ -56,7 +56,7 @@ public void iJoin(ManagementNodeInventory inv) {
     }
 
     @Override
-    public String makeDestination(String resourceUuid) {
+    public synchronized String makeDestination(String resourceUuid) {
         String nodeUuid = nodeHash.get(resourceUuid);
         if (nodeUuid == null) {
             throw new CloudRuntimeException("Cannot find any available management node to send message");
@@ -66,18 +66,18 @@ public String makeDestination(String resourceUuid) {
     }
 
     @Override
-    public boolean isManagedByUs(String resourceUuid) {
+    public synchronized boolean isManagedByUs(String resourceUuid) {
         String nodeUuid = makeDestination(resourceUuid);
         return nodeUuid.equals(Platform.getManagementServerId());
     }
 
     @Override
-    public Collection<String> getManagementNodesInHashRing() {
-        return nodeHash.getNodes();
+    public synchronized Collection<String> getManagementNodesInHashRing() {
+        return new ArrayList<>(nodeHash.getNodes());
     }
 
     @Override
-    public NodeInfo getNodeInfo(String nodeUuid) {
+    public synchronized NodeInfo getNodeInfo(String nodeUuid) {
         NodeInfo info = nodes.get(nodeUuid);
         if (info == null) {
             ManagementNodeVO vo = dbf.findByUuid(nodeUuid, ManagementNodeVO.class);
@@ -93,17 +93,17 @@ public NodeInfo getNodeInfo(String nodeUuid) {
     }
 
     @Override
-    public Collection<NodeInfo> getAllNodeInfo() {
-        return nodes.values();
+    public synchronized Collection<NodeInfo> getAllNodeInfo() {
+        return new ArrayList<>(nodes.values());
     }
 
     @Override
-    public int getManagementNodeCount() {
-        return nodes.values().size();
+    public synchronized int getManagementNodeCount() {
+        return nodes.size();
     }
 
 
-    public boolean isNodeInCircle(String nodeId) {
+    public synchronized boolean isNodeInCircle(String nodeId) {
         return nodeHash.hasNode(nodeId);
     }
 }

core/src/main/java/org/zstack/core/thread/DispatchQueueImpl.java

@@ -302,7 +302,7 @@ private class SyncTaskFuture<T> extends AbstractFuture<T> {
 
         public SyncTaskFuture(SyncTask<T> task) {
             super(task);
-            this.parentContext = Context.current();
+            this.parentContext = isTelemetryEnabled() ? Context.current() : null;
         }
 
         private SyncTask getTask() {

header/src/main/java/org/zstack/header/storage/addon/primary/ExternalPrimaryStorageInventory.java

@@ -4,8 +4,10 @@
 import org.zstack.header.storage.primary.PrimaryStorageInventory;
 import org.zstack.utils.gson.JSONObjectUtil;
 
+import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 @Inventory(mappingVOClass = ExternalPrimaryStorageVO.class)
@@ -59,6 +61,7 @@ public ExternalPrimaryStorageInventory(ExternalPrimaryStorageVO lvo) {
         super(lvo);
         identity = lvo.getIdentity();
         config = JSONObjectUtil.toObject(lvo.getConfig(), LinkedHashMap.class);
+        desensitizeConfig(config);
         addonInfo = JSONObjectUtil.toObject(lvo.getAddonInfo(), LinkedHashMap.class);
         outputProtocols = lvo.getOutputProtocols().stream().map(PrimaryStorageOutputProtocolRefVO::getOutputProtocol).collect(Collectors.toList());
         defaultProtocol = lvo.getDefaultProtocol();
@@ -68,6 +71,35 @@ public static ExternalPrimaryStorageInventory valueOf(ExternalPrimaryStorageVO l
         return new ExternalPrimaryStorageInventory(lvo);
     }
 
+    private static void desensitizeConfig(Map config) {
+        if (config == null) return;
+        desensitizeUrlList(config, "mdsUrls");
+        desensitizeUrlList(config, "mdsInfos");
+    }
+
+    private static void desensitizeUrlList(Map config, String key) {
+        Object urls = config.get(key);
+        if (urls instanceof List) {
+            List<String> desensitized = new ArrayList<>();
+            for (Object url : (List) urls) {
+                desensitized.add(desensitizeUrl(String.valueOf(url)));
+            }
+            config.put(key, desensitized);
+        }
+    }
+
+    private static String desensitizeUrl(String url) {
+        int atIndex = url.lastIndexOf('@');
+        if (atIndex > 0) {
+            int schemeIndex = url.indexOf("://");
+            if (schemeIndex >= 0 && schemeIndex < atIndex) {
+                return url.substring(0, schemeIndex + 3) + "***" + url.substring(atIndex);
+            }
+            return "***" + url.substring(atIndex);
+        }
+        return url;
+    }
+
     public String getIdentity() {
         return identity;
     }

header/src/main/java/org/zstack/header/vm/VmInstanceSpec.java

@@ -847,7 +847,9 @@ public void setBootMode(String bootMode) {
 
     public long getRootDiskAllocateSize() {
         if (rootDiskOffering == null) {
-            return this.getImageSpec().getInventory().getSize();
+            long virtualSize = this.getImageSpec().getInventory().getSize();
+            long actualSize = this.getImageSpec().getInventory().getActualSize();
+            return Math.max(virtualSize, actualSize);
         }
         return rootDiskOffering.getDiskSize();
     }

header/src/main/java/org/zstack/header/vm/VmInstanceState.java

@@ -168,6 +168,7 @@ public enum VmInstanceState {
                 new Transaction(VmInstanceStateEvent.destroyed, VmInstanceState.Destroyed),
                 new Transaction(VmInstanceStateEvent.destroying, VmInstanceState.Destroying),
                 new Transaction(VmInstanceStateEvent.running, VmInstanceState.Running),
+                new Transaction(VmInstanceStateEvent.stopped, VmInstanceState.Stopped),
                 new Transaction(VmInstanceStateEvent.expunging, VmInstanceState.Expunging)
         );
         Destroyed.transactions(

network/src/main/java/org/zstack/network/l3/L3BasicNetwork.java

@@ -1075,6 +1075,13 @@ private void handle(APIGetFreeIpMsg msg) {
             }
             limit -= freeIpInventorys.size();
         }
+
+        Set<ReservedIpRangeVO> reservedIpRanges = self.getReservedIpRanges();
+        if (reservedIpRanges != null && !reservedIpRanges.isEmpty()) {
+            freeIpInventorys.removeIf(freeIp -> reservedIpRanges.stream().anyMatch(
+                    r -> NetworkUtils.isInRange(freeIp.getIp(), r.getStartIp(), r.getEndIp())));
+        }
+
         reply.setInventories(freeIpInventorys);
 
         bus.reply(msg, reply);

plugin/ceph/src/main/java/org/zstack/storage/ceph/primary/CephPrimaryStorageBase.java

@@ -5446,7 +5446,7 @@ private void deleteSnapshotOnPrimaryStorage(final DeleteSnapshotOnPrimaryStorage
         httpCall(DELETE_SNAPSHOT_PATH, cmd, DeleteSnapshotRsp.class, new ReturnValueCompletion<DeleteSnapshotRsp>(msg) {
             @Override
             public void success(DeleteSnapshotRsp returnValue) {
-                osdHelper.releaseAvailableCapacity(msg.getSnapshot().getPrimaryStorageInstallPath(), msg.getSnapshot().getSize());
+                osdHelper.releaseAvailableCapWithRatio(msg.getSnapshot().getPrimaryStorageInstallPath(), msg.getSnapshot().getSize());
                 bus.reply(msg, reply);
                 completion.done();
             }

plugin/loadBalancer/src/main/java/org/zstack/network/service/lb/LoadBalancerApiInterceptor.java

@@ -39,6 +39,7 @@
 import org.zstack.network.service.vip.VipVO_;
 import org.zstack.tag.PatternedSystemTag;
 import org.zstack.tag.TagManager;
+import org.zstack.core.upgrade.UpgradeGlobalConfig;
 import org.zstack.utils.*;
 import org.zstack.utils.function.ForEachFunction;
 import org.zstack.utils.logging.CLogger;
@@ -152,10 +153,22 @@ public APIMessage intercept(APIMessage msg) throws ApiMessageInterceptionExcepti
             validate((APIGetCandidateVmNicsForLoadBalancerServerGroupMsg)msg);
         } else if (msg instanceof APIChangeLoadBalancerBackendServerMsg) {
             validate((APIChangeLoadBalancerBackendServerMsg)msg);
+        } else if (msg instanceof APIDeleteLoadBalancerMsg) {
+            validate((APIDeleteLoadBalancerMsg) msg);
         }
         return msg;
     }
 
+    private void validate(APIDeleteLoadBalancerMsg msg) {
+        if (UpgradeGlobalConfig.GRAYSCALE_UPGRADE.value(Boolean.class)) {
+            LoadBalancerVO lb = dbf.findByUuid(msg.getUuid(), LoadBalancerVO.class);
+            if (lb != null && lb.getType() == LoadBalancerType.SLB) {
+                throw new ApiMessageInterceptionException(argerr(
+                        "cannot delete the standalone load balancer[uuid:%s] during grayscale upgrade", msg.getUuid()));
+            }
+        }
+    }
+
     private void validate(APIDeleteAccessControlListMsg msg) {
         /*List<String> refs = Q.New(LoadBalancerListenerACLRefVO.class).select(LoadBalancerListenerACLRefVO_.listenerUuid)
                              .eq(LoadBalancerListenerACLRefVO_.aclUuid, msg.getUuid()).isNull(LoadBalancerListenerACLRefVO_.serverGroupUuid).listValues();

plugin/portForwarding/src/main/java/org/zstack/network/service/portforwarding/PortForwardingManagerImpl.java

@@ -682,10 +682,34 @@ public void fail(ErrorCode errorCode) {
     private void handle(APICreatePortForwardingRuleMsg msg) {
         final APICreatePortForwardingRuleEvent evt = new APICreatePortForwardingRuleEvent(msg.getId());
 
-        int vipPortEnd = msg.getVipPortEnd() == null ? msg.getVipPortStart() : msg.getVipPortEnd();
-        int privatePortEnd = msg.getPrivatePortEnd() == null ? msg.getPrivatePortStart() : msg.getPrivatePortEnd();
+        thdf.chainSubmit(new ChainTask(msg) {
+            @Override
+            public String getSyncSignature() {
+                return String.format("create-portforwardingrule-vip-%s", msg.getVipUuid());
+            }
 
-        VipVO vip = dbf.findByUuid(msg.getVipUuid(), VipVO.class);
+            @Override
+            public void run(SyncTaskChain chain) {
+                int vipPortEnd = msg.getVipPortEnd() == null ? msg.getVipPortStart() : msg.getVipPortEnd();
+                int privatePortEnd = msg.getPrivatePortEnd() == null ? msg.getPrivatePortStart() : msg.getPrivatePortEnd();
+
+                // re-check VIP port overlap under sync to prevent concurrent duplicate rules
+                boolean overlap = Q.New(PortForwardingRuleVO.class)
+                        .eq(PortForwardingRuleVO_.vipUuid, msg.getVipUuid())
+                        .eq(PortForwardingRuleVO_.protocolType, PortForwardingProtocolType.valueOf(msg.getProtocolType()))
+                        .lte(PortForwardingRuleVO_.vipPortStart, vipPortEnd)
+                        .gte(PortForwardingRuleVO_.vipPortEnd, msg.getVipPortStart())
+                        .isExists();
+                if (overlap) {
+                    evt.setError(operr(ORG_ZSTACK_NETWORK_SERVICE_PORTFORWARDING_10017,
+                            "vip port range[vipStartPort:%s, vipEndPort:%s] overlaps with an existing port forwarding rule on vip[uuid:%s]",
+                            msg.getVipPortStart(), vipPortEnd, msg.getVipUuid()));
+                    bus.publish(evt);
+                    chain.next();
+                    return;
+                }
+
+                VipVO vip = dbf.findByUuid(msg.getVipUuid(), VipVO.class);
         final PortForwardingRuleVO vo = new PortForwardingRuleVO();
         if (msg.getResourceUuid() != null) {
             vo.setUuid(msg.getResourceUuid());
@@ -713,8 +737,8 @@ protected void scripts() {
             }
         }.execute();
 
-        FlowChain chain = FlowChainBuilder.newShareFlowChain();
-        chain.setName("create-portforwading");
+        FlowChain flowChain = FlowChainBuilder.newShareFlowChain();
+        flowChain.setName("create-portforwading");
         VipInventory vipInventory = VipInventory.valueOf(vip);
         if (msg.getVmNicUuid() == null) {
             ModifyVipAttributesStruct struct = new ModifyVipAttributesStruct();
@@ -727,13 +751,15 @@ protected void scripts() {
                 public void success() {
                     evt.setInventory(PortForwardingRuleInventory.valueOf(vo));
                     bus.publish(evt);
+                    chain.next();
                 }
 
                 @Override
                 public void fail(ErrorCode errorCode) {
                     dbf.remove(vo);
                     evt.setError(errorCode);
                     bus.publish(evt);
+                    chain.next();
                 }
             });
 
@@ -757,20 +783,22 @@ public void fail(ErrorCode errorCode) {
                 public void success() {
                     evt.setInventory(PortForwardingRuleInventory.valueOf(vo));
                     bus.publish(evt);
+                    chain.next();
                 }
 
                 @Override
                 public void fail(ErrorCode errorCode) {
                     dbf.remove(vo);
                     evt.setError(errorCode);
                     bus.publish(evt);
+                    chain.next();
                 }
             });
 
             return;
         }
 
-        chain.then(new ShareFlow() {
+        flowChain.then(new ShareFlow() {
             @Override
             public void setup() {
                 vo.setVmNicUuid(vmNic.getUuid());
@@ -853,20 +881,29 @@ public void fail(ErrorCode errorCode) {
         });
 
 
-        chain.done(new FlowDoneHandler(msg) {
+        flowChain.done(new FlowDoneHandler(msg) {
             @Override
             public void handle(Map data) {
                 evt.setInventory(PortForwardingRuleInventory.valueOf(dbf.reload(vo)));
                 bus.publish(evt);
+                chain.next();
             }
         }).error(new FlowErrorHandler(msg) {
             @Override
             public void handle(ErrorCode errCode, Map data) {
                 dbf.remove(vo);
                 evt.setError(errCode);
                 bus.publish(evt);
+                chain.next();
             }
         }).start();
+            }
+
+            @Override
+            public String getName() {
+                return String.format("api-create-portforwardingrule-vip-%s", msg.getVipUuid());
+            }
+        });
     }
 
     private void populateExtensions() {