<feature>[zns]: add ZNS-Cloud data sync reconciliation

conf/springConfigXml/sdnController.xml

@@ -94,4 +94,6 @@
         </zstack:plugin>
     </bean>
 
+    <bean id="DirtySyncTracker" class="org.zstack.sdnController.DirtySyncTrackerImpl" />
+
 </beans>

plugin/sdnController/src/main/java/org/zstack/sdnController/DirtySyncTracker.java

@@ -0,0 +1,30 @@
+package org.zstack.sdnController;
+
+/**
+ * Tracks which SDN controllers need a full reconciliation sync.
+ *
+ * When external operations (VM NIC create/delete, migration rollback, etc.) encounter failures
+ * that may leave Cloud and SDN controller out of sync, they call {@link #markNeedSync} to flag
+ * the affected controller.
+ *
+ * The ping tracker checks this flag after consecutive successful pings and triggers a full sync
+ * when needed.
+ */
+public interface DirtySyncTracker {
+    /**
+     * Mark a controller as needing a full sync.
+     * Idempotent — calling multiple times has no additional effect.
+     */
+    void markNeedSync(String controllerUuid);
+
+    /**
+     * Check if a controller needs a sync.
+     */
+    boolean needsSync(String controllerUuid);
+
+    /**
+     * Atomically clear the needsSync flag and return its previous value.
+     * @return true if the controller was marked as needing sync, false otherwise.
+     */
+    boolean clearNeedSync(String controllerUuid);
+}

plugin/sdnController/src/main/java/org/zstack/sdnController/DirtySyncTrackerImpl.java

@@ -0,0 +1,39 @@
+package org.zstack.sdnController;
+
+import org.zstack.utils.Utils;
+import org.zstack.utils.logging.CLogger;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * In-memory implementation of DirtySyncTracker.
+ * Thread-safe via ConcurrentHashMap.
+ * Dirty state is ephemeral — lost on MN restart, which is acceptable because
+ * the ping-triggered sync provides eventual consistency.
+ */
+public class DirtySyncTrackerImpl implements DirtySyncTracker {
+    private static final CLogger logger = Utils.getLogger(DirtySyncTrackerImpl.class);
+
+    private final Map<String, Boolean> needsSyncMap = new ConcurrentHashMap<>();
+
+    @Override
+    public void markNeedSync(String controllerUuid) {
+        needsSyncMap.put(controllerUuid, Boolean.TRUE);
+        logger.debug(String.format("[sync-tracker] marked controller[uuid:%s] needsSync", controllerUuid));
+    }
+
+    @Override
+    public boolean needsSync(String controllerUuid) {
+        return Boolean.TRUE.equals(needsSyncMap.get(controllerUuid));
+    }
+
+    @Override
+    public boolean clearNeedSync(String controllerUuid) {
+        Boolean prev = needsSyncMap.remove(controllerUuid);
+        if (Boolean.TRUE.equals(prev)) {
+            logger.info(String.format("[sync-tracker] cleared needsSync for controller[uuid:%s]", controllerUuid));
+        }
+        return Boolean.TRUE.equals(prev);
+    }
+}

plugin/sdnController/src/main/java/org/zstack/sdnController/SdnControllerBase.java

@@ -79,6 +79,9 @@ public class SdnControllerBase {
 
     public SdnControllerVO self;
 
+    // Consecutive sync failure tracking (in-memory, per controller)
+    private static final java.util.Map<String, Integer> consecutiveFailCount = new java.util.concurrent.ConcurrentHashMap<>();
+
     public SdnControllerBase(SdnControllerVO self) {
         this.self = self;
     }
@@ -141,6 +144,23 @@ public void handleMessage(SdnControllerMessage msg) {
 
     private void handle(SyncSdnControllerDataMsg msg) {
         SyncSdnControllerDataReply reply = new SyncSdnControllerDataReply();
+
+        SdnControllerFactory factory = sdnMgr.getSdnControllerFactory(self.getVendorType());
+        if (!factory.isSyncEnabled()) {
+            logger.info(String.format("sync is disabled for sdn controller[uuid:%s], skip", self.getUuid()));
+            bus.reply(msg, reply);
+            return;
+        }
+
+        if (self.getStatus() != SdnControllerStatus.Connected) {
+            logger.info(String.format("sdn controller[uuid:%s] is not connected (status=%s), skip sync",
+                    self.getUuid(), self.getStatus()));
+            bus.reply(msg, reply);
+            return;
+        }
+
+        boolean dryRun = factory.isDryRun();
+
         // Run a sync chain that only syncs data, without touching connection status
         thdf.chainSubmit(new ChainTask(reply) {
             @Override
@@ -152,21 +172,34 @@ public String getSyncSignature() {
             public void run(SyncTaskChain chain) {
                 FlowChain flowChain = sdnMgr.getSyncChain(self);
                 flowChain.getData().put(SDN_CONTROLLER_UUID, self.getUuid());
+                flowChain.getData().put(SYNC_DRY_RUN, dryRun);
                 flowChain.setName(String.format("sync-sdn-controller-data-%s-%s", self.getUuid(), self.getName()));
 
-                // Start the chain; flows in factory-provided chain should perform data sync operations
                 flowChain.done(new FlowDoneHandler(msg) {
                     @Override
                     public void handle(Map data) {
+                        populateSyncReply(reply, data, dryRun);
                         bus.reply(msg, reply);
+
+                        // Reset consecutive fail count and fire completion event
+                        consecutiveFailCount.remove(self.getUuid());
+                        fireSyncCompletedEvent(reply);
+
                         chain.next();
                     }
                 }).error(new FlowErrorHandler(msg) {
                     @Override
                     public void handle(ErrorCode errCode, Map data) {
                         SyncSdnControllerDataReply r = new SyncSdnControllerDataReply();
                         r.setError(errCode);
+                        populateSyncReply(r, data, dryRun);
                         bus.reply(msg, r);
+
+                        // Increment consecutive fail count and check alarm threshold
+                        int failCount = consecutiveFailCount.merge(self.getUuid(), 1, Integer::sum);
+                        fireSyncFailedEvent(errCode, failCount);
+                        checkSyncAlarm(failCount, factory);
+
                         chain.next();
                     }
                 }).start();
@@ -179,6 +212,63 @@ public String getName() {
         });
     }
 
+    private void populateSyncReply(SyncSdnControllerDataReply reply, Map data, boolean dryRun) {
+        reply.setDryRun(dryRun);
+        if (data == null) return;
+        Object statsObj = data.get(SYNC_STATS);
+        if (statsObj == null) return;
+        try {
+            java.lang.reflect.Method getToCreateCount = statsObj.getClass().getMethod("getToCreateCount");
+            java.lang.reflect.Method getToDeleteCount = statsObj.getClass().getMethod("getToDeleteCount");
+            java.lang.reflect.Method getToUpdateCount = statsObj.getClass().getMethod("getToUpdateCount");
+            java.lang.reflect.Method getFailedCount = statsObj.getClass().getMethod("getFailedCount");
+            reply.setToCreateCount((int) getToCreateCount.invoke(statsObj));
+            reply.setToDeleteCount((int) getToDeleteCount.invoke(statsObj));
+            reply.setToUpdateCount((int) getToUpdateCount.invoke(statsObj));
+            reply.setFailedCount((int) getFailedCount.invoke(statsObj));
+        } catch (Exception e) {
+            logger.warn(String.format("failed to read sync stats: %s", e.getMessage()));
+        }
+    }
+
+    private void fireSyncCompletedEvent(SyncSdnControllerDataReply reply) {
+        SdnControllerCanonicalEvents.SdnControllerSyncCompletedData d =
+                new SdnControllerCanonicalEvents.SdnControllerSyncCompletedData();
+        d.setSdnControllerUuid(self.getUuid());
+        d.setToCreateCount(reply.getToCreateCount());
+        d.setToDeleteCount(reply.getToDeleteCount());
+        d.setToUpdateCount(reply.getToUpdateCount());
+        d.setFailedCount(reply.getFailedCount());
+        d.setDryRun(reply.isDryRun());
+        d.setScope("FULL");
+        evtf.fire(SdnControllerCanonicalEvents.SDN_CONTROLLER_SYNC_COMPLETED_PATH, d);
+    }
+
+    private void fireSyncFailedEvent(ErrorCode errCode, int failCount) {
+        SdnControllerCanonicalEvents.SdnControllerSyncFailedData d =
+                new SdnControllerCanonicalEvents.SdnControllerSyncFailedData();
+        d.setSdnControllerUuid(self.getUuid());
+        d.setErrorCode(errCode.getCode());
+        d.setErrorDetails(errCode.getDetails());
+        d.setConsecutiveFailCount(failCount);
+        evtf.fire(SdnControllerCanonicalEvents.SDN_CONTROLLER_SYNC_FAILED_PATH, d);
+    }
+
+    private void checkSyncAlarm(int failCount, SdnControllerFactory factory) {
+        int alarmThreshold = factory.getSyncFailAlarmThreshold();
+
+        if (failCount >= alarmThreshold && (failCount - alarmThreshold) % 5 == 0) {
+            logger.error(String.format("[sync-alarm] controller[uuid:%s] has %d consecutive sync failures (threshold=%d)",
+                    self.getUuid(), failCount, alarmThreshold));
+            SdnControllerCanonicalEvents.SdnControllerSyncAlarmData d =
+                    new SdnControllerCanonicalEvents.SdnControllerSyncAlarmData();
+            d.setSdnControllerUuid(self.getUuid());
+            d.setConsecutiveFailCount(failCount);
+            d.setAlarmThreshold(alarmThreshold);
+            evtf.fire(SdnControllerCanonicalEvents.SDN_CONTROLLER_SYNC_ALARM_PATH, d);
+        }
+    }
+
     public void changeSdnControllerStatus(SdnControllerStatus status) {
         if (status == self.getStatus()) {
             return;

plugin/sdnController/src/main/java/org/zstack/sdnController/SdnControllerFactory.java

@@ -26,4 +26,12 @@ public interface SdnControllerFactory {
     default SdnControllerDhcp getSdnControllerDhcp(String l2NetworkUuid) {return null;};
 
     default FlowChain getSyncChain() {return FlowChainBuilder.newSimpleFlowChain();};
+
+    default boolean isSyncEnabled() { return false; }
+
+    default boolean isDryRun() { return false; }
+
+    default int getSyncFailAlarmThreshold() { return 3; }
+
+    default int getSyncPingSuccessThreshold() { return 10; }
 }

plugin/sdnController/src/main/java/org/zstack/sdnController/SdnControllerManagerImpl.java

@@ -66,6 +66,8 @@ public class SdnControllerManagerImpl extends AbstractService implements SdnCont
     private SecurityGroupManager sgMgr;
     @Autowired
     private SdnControllerPingTracker pingTracker;
+    @Autowired
+    private DirtySyncTracker dirtySyncTracker;
 
     private Map<String, SdnControllerFactory> sdnControllerFactories = Collections.synchronizedMap(new HashMap<String, SdnControllerFactory>());
 
@@ -405,6 +407,8 @@ public void success() {
 
                 @Override
                 public void fail(ErrorCode errorCode) {
+                    // Mark controller needsSync — port creation failed, ZNS may be out of sync
+                    dirtySyncTracker.markNeedSync(e.getKey());
                     wcomp.addError(errorCode);
                     wcomp.allDone();
                 }
@@ -443,6 +447,8 @@ public void success() {
 
                 @Override
                 public void fail(ErrorCode errorCode) {
+                    // Mark controller needsSync — port deletion failed, ZNS may be out of sync
+                    dirtySyncTracker.markNeedSync(e.getKey());
                     wcomp.addError(errorCode);
                     wcomp.allDone();
                 }

plugin/sdnController/src/main/java/org/zstack/sdnController/SdnControllerPingTracker.java

@@ -1,6 +1,7 @@
 package org.zstack.sdnController;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.zstack.core.cloudbus.CloudBusCallBack;
 import org.zstack.core.cloudbus.ResourceDestinationMaker;
 import org.zstack.core.componentloader.PluginRegistry;
 import org.zstack.core.config.GlobalConfig;
@@ -27,6 +28,9 @@
 import org.zstack.utils.logging.CLogger;
 
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Created by shixin on 06/26/2025.
@@ -43,6 +47,10 @@ public class SdnControllerPingTracker extends PingTracker implements
     protected PluginRegistry pluginRgty;
     @Autowired
     protected SdnControllerManager sdnMgr;
+    @Autowired
+    private DirtySyncTracker dirtySyncTracker;
+
+    private final Map<String, AtomicInteger> consecutivePingSuccessCount = new ConcurrentHashMap<>();
 
     public String getResourceName() {
         return "sdn controller";
@@ -81,23 +89,59 @@ public void handleReply(final String resourceUuid, MessageReply reply) {
             return;
         }
 
-
         if (!reply.isSuccess()) {
             logger.warn(String.format("[SDN Ping Tracker]: unable to ping the sdn controller[uuid: %s], %s", resourceUuid, reply.getError()));
+            consecutivePingSuccessCount.remove(resourceUuid);
             new SdnControllerBase(vo).changeSdnControllerStatus(SdnControllerStatus.Disconnected);
             return;
         }
 
         SdnControllerStatus oldStatus = vo.getStatus();
         if (oldStatus == SdnControllerStatus.Disconnected) {
             // when reconnect successfully, it will fire event: SdnControllerStatus.Connected
+            consecutivePingSuccessCount.remove(resourceUuid);
             ReconnectSdnControllerMsg msg = new ReconnectSdnControllerMsg();
             msg.setControllerUuid(resourceUuid);
             bus.makeTargetServiceIdByResourceUuid(msg, SdnControllerConstant.SERVICE_ID, resourceUuid);
             bus.send(msg);
+            return;
+        }
+
+        // Ping success on Connected controller: count consecutive successes and trigger sync if needed
+        if (vo.getStatus() == SdnControllerStatus.Connected) {
+            int count = consecutivePingSuccessCount
+                    .computeIfAbsent(resourceUuid, k -> new AtomicInteger(0))
+                    .incrementAndGet();
+
+            SdnControllerFactory factory = sdnMgr.getSdnControllerFactory(vo.getVendorType());
+            if (factory.isSyncEnabled() && dirtySyncTracker.needsSync(resourceUuid)) {
+                int threshold = factory.getSyncPingSuccessThreshold();
+                if (count >= threshold) {
+                    consecutivePingSuccessCount.remove(resourceUuid);
+                    dirtySyncTracker.clearNeedSync(resourceUuid);
+                    triggerSync(resourceUuid);
+                }
+            }
         }
     }
 
+    private void triggerSync(String controllerUuid) {
+        logger.info(String.format("[SDN Ping Tracker]: triggering sync for controller[uuid:%s] after consecutive ping successes", controllerUuid));
+        SyncSdnControllerDataMsg msg = new SyncSdnControllerDataMsg();
+        msg.setControllerUuid(controllerUuid);
+        bus.makeTargetServiceIdByResourceUuid(msg, SdnControllerConstant.SERVICE_ID, controllerUuid);
+        bus.send(msg, new CloudBusCallBack(null) {
+            @Override
+            public void run(MessageReply reply) {
+                if (!reply.isSuccess()) {
+                    logger.warn(String.format("[SDN Ping Tracker]: sync failed for controller[uuid:%s], re-marking needsSync: %s",
+                            controllerUuid, reply.getError()));
+                    dirtySyncTracker.markNeedSync(controllerUuid);
+                }
+            }
+        });
+    }
+
     private void trackOurs() {
         List<String> sdnControllerUuids = Q.New(SdnControllerVO.class)
                 .select(SdnControllerVO_.uuid).listValues();
@@ -154,6 +198,7 @@ protected void trackHook(String resourceUuid) {
 
     @Override
     public void deleteNetworkServiceOfSdnController(String sdnControllerUuid, Completion completion) {
+        consecutivePingSuccessCount.remove(sdnControllerUuid);
         super.untrack(sdnControllerUuid);
         completion.success();
     }