KVM: Propagating changes on host parameters to the agents

Author: ustcweizhou

engine/orchestration/src/com/cloud/agent/manager/AgentManagerImpl.java

@@ -46,6 +46,8 @@
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
 import org.apache.cloudstack.framework.jobs.AsyncJob;
 import org.apache.cloudstack.framework.jobs.AsyncJobExecutionContext;
+import org.apache.cloudstack.framework.messagebus.MessageBus;
+import org.apache.cloudstack.framework.messagebus.MessageSubscriber;
 import org.apache.cloudstack.managed.context.ManagedContextRunnable;
 import org.apache.cloudstack.outofbandmanagement.dao.OutOfBandManagementDao;
 import org.apache.cloudstack.utils.identity.ManagementServerNode;
@@ -83,6 +85,7 @@
 import com.cloud.dc.dao.ClusterDao;
 import com.cloud.dc.dao.DataCenterDao;
 import com.cloud.dc.dao.HostPodDao;
+import com.cloud.event.EventTypes;
 import com.cloud.exception.AgentUnavailableException;
 import com.cloud.exception.ConnectionException;
 import com.cloud.exception.OperationTimedoutException;
@@ -189,6 +192,8 @@ public class AgentManagerImpl extends ManagerBase implements AgentManager, Handl
 
     @Inject
     ManagementServiceConfiguration mgmtServiceConf;
+    @Inject
+    MessageBus messageBus;
 
     protected final ConfigKey<Integer> Workers = new ConfigKey<Integer>("Advanced", Integer.class, "workers", "5",
                     "Number of worker threads handling remote agent connections.", false);
@@ -244,6 +249,8 @@ public boolean configure(final String name, final Map<String, Object> params) th
 
         _monitorExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("AgentMonitor"));
 
+        initMessageBusListener();
+
         return true;
     }
 
@@ -1803,4 +1810,57 @@ public int getTimeout() {
 
     }
 
+    protected Map<Long, List<Long>> getHostsPerZone() {
+        List<HostVO> allHosts = _resourceMgr.listAllHostsInAllZonesByType(Host.Type.Routing);
+        if (allHosts == null) {
+            return null;
+        }
+        Map<Long, List<Long>> hostsByZone = new HashMap<Long, List<Long>>();
+        for (HostVO host : allHosts) {
+            if (host.getHypervisorType() == HypervisorType.KVM || host.getHypervisorType() == HypervisorType.LXC) {
+                Long zoneId = host.getDataCenterId();
+                List<Long> hostIds = hostsByZone.get(zoneId);
+                if (hostIds == null) {
+                    hostIds = new ArrayList<Long>();
+                }
+                hostIds.add(host.getId());
+                hostsByZone.put(zoneId, hostIds);
+            }
+        }
+        return hostsByZone;
+    }
+
+    private void sendCommandToAgents(Map<Long, List<Long>> hostsPerZone, Map<String, String> params) {
+        SetHostParamsCommand cmds = new SetHostParamsCommand(params);
+        for (Long zoneId : hostsPerZone.keySet()) {
+            List<Long> hostIds = hostsPerZone.get(zoneId);
+            for (Long hostId : hostIds) {
+                Answer answer = easySend(hostId, cmds);
+                if (answer == null || !answer.getResult()) {
+                    s_logger.error("Error sending parameters to agent " + hostId);
+                }
+            }
+        }
+    }
+
+    private void propagateChangeToAgents() {
+        s_logger.debug("Propagating changes on host parameters to the agents");
+        Map<Long, List<Long>> hostsPerZone = getHostsPerZone();
+        Map<String, String> params = new HashMap<String, String>();
+        params.put("router.aggregation.command.each.timeout", _configDao.getValue("router.aggregation.command.each.timeout"));
+        sendCommandToAgents(hostsPerZone, params);
+    }
+
+    private void initMessageBusListener() {
+        messageBus.subscribe(EventTypes.EVENT_CONFIGURATION_VALUE_EDIT, new MessageSubscriber() {
+            @Override
+            public void onPublishMessage(String serderAddress, String subject, Object args) {
+                String globalSettingUpdated = (String)args;
+                if (globalSettingUpdated.equals("router.aggregation.command.each.timeout")) {
+                    propagateChangeToAgents();
+                }
+            }
+        });
+    }
+
 }

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java

@@ -102,6 +102,7 @@
 import com.cloud.agent.api.to.NfsTO;
 import com.cloud.agent.api.to.NicTO;
 import com.cloud.agent.api.to.VirtualMachineTO;
+import com.cloud.agent.dao.impl.PropertiesStorage;
 import com.cloud.agent.resource.virtualnetwork.VRScripts;
 import com.cloud.agent.resource.virtualnetwork.VirtualRouterDeployer;
 import com.cloud.agent.resource.virtualnetwork.VirtualRoutingResource;
@@ -1066,6 +1067,24 @@ public boolean configure(final String name, final Map<String, Object> params) th
         return true;
     }
 
+    public boolean configureHostParams(final Map<String, String> params) {
+        final File file = PropertiesUtil.findConfigFile("agent.properties");
+        if (file == null) {
+            s_logger.error("Unable to find the file agent.properties");
+            return false;
+        }
+        // Save configurations in agent.properties
+        PropertiesStorage storage = new PropertiesStorage();
+        storage.configure("Storage", new HashMap<String, Object>());
+        if (params.get("router.aggregation.command.each.timeout") == null) {
+            String value = (String)params.get("router.aggregation.command.each.timeout");
+            Integer intValue = NumbersUtil.parseInt(value, 10);
+            storage.persist("router.aggregation.command.each.timeout", String.valueOf(intValue));
+        }
+
+        return true;
+    }
+
     protected void configureDiskActivityChecks(final Map<String, Object> params) {
         _diskActivityCheckEnabled = Boolean.parseBoolean((String)params.get("vm.diskactivity.checkenabled"));
         if (_diskActivityCheckEnabled) {

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtSetHostParamsCommandWrapper.java

@@ -35,6 +35,7 @@ public Answer execute(final SetHostParamsCommand command, final LibvirtComputing
 
         final Map<String, String> params = command.getParams();
         boolean success = libvirtComputingResource.getVirtRouterResource().configureHostParams(params);
+        success = success && libvirtComputingResource.configureHostParams(params);
 
         if (!success) {
             return new Answer(command, false, "Failed to set host parameters");