wip changes for extension sync

shwstppr · shwstppr · commit 17eb17cd0ac6 · 2025-06-10T10:24:02.000+05:30
Signed-off-by: Abhishek Kumar &lt;abhishek.mrt22@gmail.com&gt;
diff --git a/api/src/main/java/org/apache/cloudstack/alert/AlertService.java b/api/src/main/java/org/apache/cloudstack/alert/AlertService.java
@@ -73,6 +73,7 @@ private AlertType(short type, String name, boolean isDefault) {
         public static final AlertType ALERT_TYPE_VM_SNAPSHOT = new AlertType((short)32, "ALERT.VM.SNAPSHOT", true);
         public static final AlertType ALERT_TYPE_VR_PUBLIC_IFACE_MTU = new AlertType((short)32, "ALERT.VR.PUBLIC.IFACE.MTU", true);
         public static final AlertType ALERT_TYPE_VR_PRIVATE_IFACE_MTU = new AlertType((short)32, "ALERT.VR.PRIVATE.IFACE.MTU", true);
+        public static final AlertType ALERT_TYPE_EXTENSION_ENTRY_POINT_OUT_OF_SYNC = new AlertType((short)33, "ALERT.TYPE.EXTENSION.ENTRY.POINT.OUT.OF.SYNC", true);
 
         public short getType() {
             return type;
diff --git a/api/src/main/java/org/apache/cloudstack/extension/Extension.java b/api/src/main/java/org/apache/cloudstack/extension/Extension.java
@@ -30,6 +30,7 @@ enum Type {
     String getDescription();
     Type getType();
     String getRelativeEntryPoint();
+    boolean isEntryPointSync();
     boolean isUserDefined();
     Date getCreated();
 }
diff --git a/engine/components-api/src/main/java/com/cloud/hypervisor/ExternalProvisioner.java b/engine/components-api/src/main/java/com/cloud/hypervisor/ExternalProvisioner.java
@@ -50,6 +50,8 @@ public interface ExternalProvisioner extends Manager {
 
     String getExtensionEntryPoint(String relativeEntryPoint);
 
+    String getChecksumForExtensionEntryPoint(String extensionName, String relativeEntryPoint);
+
     void prepareExtensionEntryPoint(String extensionName, boolean userDefined, String extensionRelativeEntryPoint);
 
     PrepareExternalProvisioningAnswer prepareExternalProvisioning(String hostGuid, String extensionName, String extensionRelativeEntryPoint, PrepareExternalProvisioningCommand cmd);
diff --git a/engine/orchestration/src/main/java/com/cloud/agent/manager/ClusteredAgentManagerImpl.java b/engine/orchestration/src/main/java/com/cloud/agent/manager/ClusteredAgentManagerImpl.java
@@ -46,6 +46,8 @@
 import org.apache.cloudstack.framework.config.ConfigDepot;
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
+import org.apache.cloudstack.framework.extensions.command.GetExtensionEntryPointChecksumCommand;
+import org.apache.cloudstack.framework.extensions.manager.ExtensionsManager;
 import org.apache.cloudstack.ha.dao.HAConfigDao;
 import org.apache.cloudstack.maintenance.ManagementServerMaintenanceManager;
 import org.apache.cloudstack.maintenance.command.BaseShutdownManagementServerHostCommand;
@@ -60,6 +62,7 @@
 import org.apache.cloudstack.outofbandmanagement.dao.OutOfBandManagementDao;
 import org.apache.cloudstack.utils.identity.ManagementServerNode;
 import org.apache.cloudstack.utils.security.SSLUtils;
+import org.apache.commons.collections.CollectionUtils;
 
 import com.cloud.agent.api.Answer;
 import com.cloud.agent.api.CancelCommand;
@@ -105,8 +108,6 @@
 import com.cloud.utils.nio.Task;
 import com.google.gson.Gson;
 
-import org.apache.commons.collections.CollectionUtils;
-
 public class ClusteredAgentManagerImpl extends AgentManagerImpl implements ClusterManagerListener, ClusteredAgentRebalanceService {
     private static ScheduledExecutorService s_transferExecutor = Executors.newScheduledThreadPool(2, new NamedThreadFactory("Cluster-AgentRebalancingExecutor"));
     private final long rebalanceTimeOut = 300000; // 5 mins - after this time remove the agent from the transfer list
@@ -146,6 +147,8 @@ public class ClusteredAgentManagerImpl extends AgentManagerImpl implements Clust
     private ManagementServerMaintenanceManager managementServerMaintenanceManager;
     @Inject
     private DataCenterDao dcDao;
+    @Inject
+    ExtensionsManager extensionsManager;
 
     protected ClusteredAgentManagerImpl() {
         super();
@@ -1321,6 +1324,9 @@ public String dispatch(final ClusterServicePdu pdu) {
             } else if (cmds.length == 1 && cmds[0] instanceof BaseShutdownManagementServerHostCommand) {
                 final BaseShutdownManagementServerHostCommand cmd = (BaseShutdownManagementServerHostCommand) cmds[0];
                 return handleShutdownManagementServerHostCommand(cmd);
+            } else if (cmds.length == 1 && cmds[0] instanceof GetExtensionEntryPointChecksumCommand) {
+                final GetExtensionEntryPointChecksumCommand cmd = (GetExtensionEntryPointChecksumCommand) cmds[0];
+                return extensionsManager.handleGetExtensionEntryPointChecksumCommand(cmd);
             }
 
             try {
diff --git a/engine/schema/src/main/resources/META-INF/db/schema-42010to42100.sql b/engine/schema/src/main/resources/META-INF/db/schema-42010to42100.sql
@@ -91,6 +91,7 @@ CREATE TABLE IF NOT EXISTS `cloud`.`extension` (
   `description` varchar(4096),
   `type` varchar(255) NOT NULL COMMENT 'Type of the extension: Orchestrator, etc',
   `relative_entry_point` varchar(2048) NOT NULL COMMENT 'Path of entry point for the extension relative to the root extensions directory',
+  `entry_point_sync` int unsigned DEFAULT '0' COMMENT 'True if the extension entry point is in sync across management servers',
   `is_user_defined` int unsigned DEFAULT '0' COMMENT 'True if the extension is added by admin',
   `created` datetime NOT NULL,
   `removed` datetime DEFAULT NULL,
diff --git a/framework/extensions/src/main/java/org/apache/cloudstack/framework/extensions/command/GetExtensionEntryPointChecksumCommand.java b/framework/extensions/src/main/java/org/apache/cloudstack/framework/extensions/command/GetExtensionEntryPointChecksumCommand.java
@@ -0,0 +1,56 @@
+// 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.cloudstack.framework.extensions.command;
+
+import com.cloud.agent.api.Command;
+
+public class GetExtensionEntryPointChecksumCommand extends Command {
+    long msId;
+    long extensionId;
+    String extensionName;
+    String extensionRelativeEntryPointPath;
+
+    public GetExtensionEntryPointChecksumCommand(long msId, long extensionId, String extensionName,
+                             String extensionRelativeEntryPointPath) {
+        this.msId = msId;
+        this.extensionId = extensionId;
+        this.extensionName = extensionName;
+        this.extensionRelativeEntryPointPath = extensionRelativeEntryPointPath;
+    }
+
+    public long getMsId() {
+        return msId;
+    }
+
+    public long getExtensionId() {
+        return extensionId;
+    }
+
+    public String getExtensionName() {
+        return extensionName;
+    }
+
+    public String getExtensionRelativeEntryPointPath() {
+        return extensionRelativeEntryPointPath;
+    }
+
+    @Override
+    public boolean executeInSequence() {
+        return false;
+    }
+}
diff --git a/framework/extensions/src/main/java/org/apache/cloudstack/framework/extensions/manager/ExtensionsManager.java b/framework/extensions/src/main/java/org/apache/cloudstack/framework/extensions/manager/ExtensionsManager.java
@@ -42,6 +42,7 @@
 import org.apache.cloudstack.framework.extensions.api.UnregisterExtensionCmd;
 import org.apache.cloudstack.framework.extensions.api.UpdateCustomActionCmd;
 import org.apache.cloudstack.framework.extensions.api.UpdateExtensionCmd;
+import org.apache.cloudstack.framework.extensions.command.GetExtensionEntryPointChecksumCommand;
 
 import com.cloud.host.Host;
 import com.cloud.org.Cluster;
@@ -80,4 +81,6 @@ public interface ExtensionsManager extends Manager {
     ExtensionCustomActionResponse createCustomActionResponse(ExtensionCustomAction customAction);
 
     Map<String, Object> getExternalAccessDetails(Host host);
+
+    String handleGetExtensionEntryPointChecksumCommand(GetExtensionEntryPointChecksumCommand cmd);
 }
diff --git a/framework/extensions/src/main/java/org/apache/cloudstack/framework/extensions/manager/ExtensionsManagerImpl.java b/framework/extensions/src/main/java/org/apache/cloudstack/framework/extensions/manager/ExtensionsManagerImpl.java
@@ -26,16 +26,21 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import javax.inject.Inject;
+import javax.naming.ConfigurationException;
 
 import org.apache.cloudstack.acl.RoleType;
 import org.apache.cloudstack.api.ApiConstants;
@@ -59,6 +64,7 @@
 import org.apache.cloudstack.framework.extensions.api.UnregisterExtensionCmd;
 import org.apache.cloudstack.framework.extensions.api.UpdateCustomActionCmd;
 import org.apache.cloudstack.framework.extensions.api.UpdateExtensionCmd;
+import org.apache.cloudstack.framework.extensions.command.GetExtensionEntryPointChecksumCommand;
 import org.apache.cloudstack.framework.extensions.dao.ExtensionCustomActionDao;
 import org.apache.cloudstack.framework.extensions.dao.ExtensionCustomActionDetailsDao;
 import org.apache.cloudstack.framework.extensions.dao.ExtensionDao;
@@ -71,6 +77,9 @@
 import org.apache.cloudstack.framework.extensions.vo.ExtensionResourceMapDetailsVO;
 import org.apache.cloudstack.framework.extensions.vo.ExtensionResourceMapVO;
 import org.apache.cloudstack.framework.extensions.vo.ExtensionVO;
+import org.apache.cloudstack.managed.context.ManagedContextRunnable;
+import org.apache.cloudstack.management.ManagementServerHost;
+import org.apache.cloudstack.utils.identity.ManagementServerNode;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang3.EnumUtils;
@@ -79,8 +88,13 @@
 
 import com.cloud.agent.AgentManager;
 import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.Command;
 import com.cloud.agent.api.RunCustomActionAnswer;
 import com.cloud.agent.api.RunCustomActionCommand;
+import com.cloud.alert.AlertManager;
+import com.cloud.cluster.ClusterManager;
+import com.cloud.cluster.ManagementServerHostVO;
+import com.cloud.cluster.dao.ManagementServerHostDao;
 import com.cloud.dc.ClusterVO;
 import com.cloud.dc.dao.ClusterDao;
 import com.cloud.event.ActionEvent;
@@ -95,11 +109,14 @@
 import com.cloud.hypervisor.ExternalProvisioner;
 import com.cloud.hypervisor.Hypervisor;
 import com.cloud.org.Cluster;
+import com.cloud.serializer.GsonHelper;
 import com.cloud.utils.Pair;
 import com.cloud.utils.component.ManagerBase;
 import com.cloud.utils.component.PluggableService;
+import com.cloud.utils.concurrency.NamedThreadFactory;
 import com.cloud.utils.db.EntityManager;
 import com.cloud.utils.db.Filter;
+import com.cloud.utils.db.GlobalLock;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.Transaction;
@@ -154,6 +171,17 @@ public class ExtensionsManagerImpl extends ManagerBase implements ExtensionsMana
     @Inject
     EntityManager entityManager;
 
+    @Inject
+    ManagementServerHostDao managementServerHostDao;
+
+    @Inject
+    ClusterManager clusterManager;
+
+    @Inject
+    AlertManager alertManager;
+
+    private ScheduledExecutorService entryPointSyncCheckExecutor;
+
     protected String getExtensionSafeName(String name) {
         return  name.replaceAll("[^a-zA-Z0-9._-]", "_").toLowerCase();
     }
@@ -969,6 +997,37 @@ private Map<String, Object> getExternalAccessDetails(Map<String, String> actionD
         return externalDetails;
     }
 
+    @Override
+    public String handleGetExtensionEntryPointChecksumCommand(GetExtensionEntryPointChecksumCommand cmd) {
+        final String extensionName = cmd.getExtensionName();
+        final String extensionRelativeEntryPointPath = cmd.getExtensionRelativeEntryPointPath();
+        logger.debug("Received GetExtensionEntryPointChecksumCommand from MS: {} for extension [id: {}, name: {}, relativeEntryPoint: {}]",
+                cmd.getMsId(), cmd.getExtensionId(), extensionName, extensionRelativeEntryPointPath);
+        return externalProvisioner.getChecksumForExtensionEntryPoint(extensionName, extensionRelativeEntryPointPath);
+    }
+
+    @Override
+    public boolean start() {
+        long syncCheckInitialDelay = 120;
+        long syncCheckInterval = 600;
+        logger.debug("Scheduling extensions entrypoint sync check task with initial delay={}s and interval={}s",
+                syncCheckInitialDelay, syncCheckInterval);
+        entryPointSyncCheckExecutor.scheduleWithFixedDelay(new EntryPointSyncCheckWorker(),
+                syncCheckInitialDelay, syncCheckInterval, TimeUnit.SECONDS);
+        return true;
+    }
+
+    @Override
+    public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
+        try {
+            entryPointSyncCheckExecutor = Executors.newScheduledThreadPool(1,
+                    new NamedThreadFactory("Extension-EntryPoint-Sync-Check"));
+        } catch (final Exception e) {
+            throw new ConfigurationException("Unable to to configure ExtensionsManagerImpl");
+        }
+        return true;
+    }
+
     @Override
     public List<Class<?>> getCommands() {
         List<Class<?>> cmds = new ArrayList<>();
@@ -986,4 +1045,94 @@ public List<Class<?>> getCommands() {
         cmds.add(UnregisterExtensionCmd.class);
         return cmds;
     }
+
+    public class EntryPointSyncCheckWorker extends ManagedContextRunnable {
+        protected void sendExtensionEntryPointOutOfSyncAlert(Extension extension) {
+            String msg = String.format("Entry-point for %s are out of sync across management servers",
+                    extension);
+            alertManager.sendAlert(AlertManager.AlertType.ALERT_TYPE_USERVM, 0L, 0L, msg, msg);
+        }
+
+        protected void updateExtensionSync(Extension extension, boolean sync) {
+            if (!sync) {
+                sendExtensionEntryPointOutOfSyncAlert(extension);
+            }
+            if (extension.isEntryPointSync() == sync) {
+                return;
+            }
+            ExtensionVO extensionVO = extensionDao.createForUpdate(extension.getId());
+            extensionVO.setEntryPointSync(sync);
+            extensionDao.update(extension.getId(), extensionVO);
+        }
+
+        protected String getChecksumFromMSPeer(Extension extension, ManagementServerHostVO msHost) {
+            final String msPeer = Long.toString(msHost.getMsid());
+            logger.debug("Retrieving checksum for {} from MS: {}", extension, msPeer);
+            final Command[] cmds = new Command[1];
+            cmds[0] = new GetExtensionEntryPointChecksumCommand(ManagementServerNode.getManagementServerId(),
+                    extension.getId(), extension.getName(), extension.getRelativeEntryPoint());
+            return clusterManager.execute(msPeer, 0L, GsonHelper.getGson().toJson(cmds), true);
+        }
+
+        protected void checkSyncForOrchestrator(Extension extension, List<ManagementServerHostVO> msHosts) {
+            if (CollectionUtils.isEmpty(msHosts)) {
+                updateExtensionSync(extension, true);
+                return;
+            }
+            String checksum = externalProvisioner.getChecksumForExtensionEntryPoint(extension.getName(),
+                    extension.getRelativeEntryPoint());
+            if (StringUtils.isBlank(checksum)) {
+                updateExtensionSync(extension, false);
+                return;
+            }
+            for (ManagementServerHostVO msHost : msHosts) {
+                final String msPeerChecksum = getChecksumFromMSPeer(extension, msHost);
+                if (!checksum.equals(msPeerChecksum)) {
+                    logger.warn("Entry-point checksum for {} is different [msid: {}, checksum: {}] and [msid: {}, checksum: {}]",
+                            extension, ManagementServerNode.getManagementServerId(), checksum,
+                            msHost.getMsid(), msPeerChecksum);
+                    updateExtensionSync(extension, false);
+                    return;
+                }
+            }
+            updateExtensionSync(extension, true);
+        }
+
+        protected void runCleanupForLongestRunningManagementServer() {
+            try {
+                List<ManagementServerHostVO> msHosts = managementServerHostDao.listBy(ManagementServerHost.State.Up);
+                msHosts.sort(Comparator.comparingLong(ManagementServerHostVO::getRunid));
+                ManagementServerHostVO msHost = msHosts.remove(0);
+                if (msHost == null || (msHost.getMsid() != ManagementServerNode.getManagementServerId())) {
+                    logger.debug("Skipping the extensions entrypoint sync check on this management server");
+                    return;
+                }
+                List<ExtensionVO> extensions = extensionDao.listAll();
+                for (ExtensionVO extension : extensions) {
+                    if (!Extension.Type.Orchestrator.equals(extension.getType())) {
+                        continue;
+                    }
+                    checkSyncForOrchestrator(extension, msHosts);
+                }
+            } catch (Exception e) {
+                logger.warn("Cleanup task failed to cleanup old webhook deliveries", e);
+            }
+        }
+
+        @Override
+        protected void runInContext() {
+            GlobalLock gcLock = GlobalLock.getInternLock("ExtensionEntryPointSyncCheck");
+            try {
+                if (gcLock.lock(3)) {
+                    try {
+                        runCleanupForLongestRunningManagementServer();
+                    } finally {
+                        gcLock.unlock();
+                    }
+                }
+            } finally {
+                gcLock.releaseRef();
+            }
+        }
+    }
 }
diff --git a/framework/extensions/src/main/java/org/apache/cloudstack/framework/extensions/vo/ExtensionVO.java b/framework/extensions/src/main/java/org/apache/cloudstack/framework/extensions/vo/ExtensionVO.java
@@ -76,6 +76,9 @@ public ExtensionVO(String name, String description, Type type, String relativeEn
     @Column(name = "relative_entry_point", nullable = false, length = 2048)
     private String relativeEntryPoint;
 
+    @Column(name = "entry_point_sync")
+    private boolean entryPointSync;
+
     @Column(name = "is_user_defined")
     private boolean userDefined;
 
@@ -127,6 +130,15 @@ public void setRelativeEntryPoint(String relativeEntryPoint) {
         this.relativeEntryPoint = relativeEntryPoint;
     }
 
+    @Override
+    public boolean isEntryPointSync() {
+        return entryPointSync;
+    }
+
+    public void setEntryPointSync(boolean entryPointSync) {
+        this.entryPointSync = entryPointSync;
+    }
+
     @Override
     public boolean isUserDefined() {
         return userDefined;
diff --git a/plugins/hypervisors/external/src/main/java/org/apache/cloudstack/hypervisor/external/provisioner/simpleprovisioner/SimpleExternalProvisioner.java b/plugins/hypervisors/external/src/main/java/org/apache/cloudstack/hypervisor/external/provisioner/simpleprovisioner/SimpleExternalProvisioner.java

Original file line number	Diff line number	Diff line change
`@@ -30,6 +30,7 @@ enum Type {`
`30`	`30`	`String getDescription();`
`31`	`31`	`Type getType();`
`32`	`32`	`String getRelativeEntryPoint();`
	`33`	`+ boolean isEntryPointSync();`
`33`	`34`	`boolean isUserDefined();`
`34`	`35`	`Date getCreated();`
`35`	`36`	`}`