CLOUDSTACK-9569: propagate global configuration router.aggregation.command.each.timeout to KVM agent

Author: ustcweizhou

core/src/com/cloud/agent/api/SetHostParamsCommand.java

@@ -0,0 +1,43 @@
+//
+// 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 com.cloud.agent.api;
+
+import java.util.Map;
+
+public class SetHostParamsCommand extends Command {
+
+    Map<String, String> params;
+
+    public SetHostParamsCommand(Map<String, String> params) {
+        this.params = params;
+    }
+
+    public Map<String, String> getParams() {
+        return params;
+    }
+
+    protected SetHostParamsCommand() {
+    }
+
+    @Override
+    public boolean executeInSequence() {
+        return true;
+    }
+}

core/src/com/cloud/agent/resource/virtualnetwork/VirtualRoutingResource.java

@@ -75,6 +75,7 @@ public class VirtualRoutingResource {
     private int _retry;
     private int _port;
     private Duration _eachTimeout;
+    private Map<String, Object> _params;
 
     private String _cfgVersion = "1.0";
 
@@ -259,8 +260,18 @@ private Answer execute(GetDomRVersionCmd cmd) {
         return new GetDomRVersionAnswer(cmd, result.getDetails(), lines[0], lines[1]);
     }
 
+    public boolean configureHostParams(final Map<String, String> params) {
+        if (_params.get("router.aggregation.command.each.timeout") == null) {
+            String value = (String)params.get("router.aggregation.command.each.timeout");
+            _eachTimeout = Duration.standardSeconds(NumbersUtil.parseInt(value, 10));
+        }
+
+        return true;
+    }
+
     public boolean configure(final String name, final Map<String, Object> params) throws ConfigurationException {
         _name = name;
+        _params = params;
 
         String value = (String)params.get("ssh.sleep");
         _sleep = NumbersUtil.parseInt(value, 10) * 1000;

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

@@ -61,6 +61,7 @@
 import com.cloud.agent.api.PingRoutingCommand;
 import com.cloud.agent.api.ReadyAnswer;
 import com.cloud.agent.api.ReadyCommand;
+import com.cloud.agent.api.SetHostParamsCommand;
 import com.cloud.agent.api.ShutdownCommand;
 import com.cloud.agent.api.StartupAnswer;
 import com.cloud.agent.api.StartupCommand;
@@ -214,6 +215,8 @@ public boolean configure(final String name, final Map<String, Object> params) th
 
         registerForHostEvents(new BehindOnPingListener(), true, true, false);
 
+        registerForHostEvents(new SetHostParamsListener(), true, true, false);
+
         _executor = new ThreadPoolExecutor(threads, threads, 60l, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("AgentTaskPool"));
 
         _connectExecutor = new ThreadPoolExecutor(100, 500, 60l, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("AgentConnectTaskPool"));
@@ -1710,4 +1713,73 @@ public ConfigKey<?>[] getConfigKeys() {
                         DirectAgentThreadCap };
     }
 
+    protected class SetHostParamsListener implements Listener {
+        @Override
+        public boolean isRecurring() {
+            return false;
+        }
+
+        @Override
+        public boolean processAnswers(final long agentId, final long seq, final Answer[] answers) {
+            return false;
+        }
+
+        @Override
+        public boolean processCommands(final long agentId, final long seq, final Command[] commands) {
+            return false;
+        }
+
+        @Override
+        public AgentControlAnswer processControlCommand(final long agentId, final AgentControlCommand cmd) {
+            return null;
+        }
+
+        @Override
+        public void processHostAdded(long hostId) {
+        }
+
+        @Override
+        public void processConnect(final Host host, final StartupCommand cmd, final boolean forRebalance) {
+        if (cmd instanceof StartupRoutingCommand) {
+            if (((StartupRoutingCommand)cmd).getHypervisorType() == HypervisorType.KVM || ((StartupRoutingCommand)cmd).getHypervisorType() == HypervisorType.LXC) {
+                Map<String, String> params = new HashMap<String, String>();
+                params.put("router.aggregation.command.each.timeout", _configDao.getValue("router.aggregation.command.each.timeout"));
+
+                try {
+                    SetHostParamsCommand cmds = new SetHostParamsCommand(params);
+                    Commands c = new Commands(cmds);
+                    send(host.getId(), c, this);
+                } catch (AgentUnavailableException e) {
+                    s_logger.debug("Failed to send host params on host: " + host.getId());
+                }
+            }
+        }
+
+        }
+
+        @Override
+        public boolean processDisconnect(final long agentId, final Status state) {
+            return true;
+        }
+
+        @Override
+        public void processHostAboutToBeRemoved(long hostId) {
+        }
+
+        @Override
+        public void processHostRemoved(long hostId, long clusterId) {
+        }
+
+        @Override
+        public boolean processTimeout(final long agentId, final long seq) {
+            return false;
+        }
+
+        @Override
+        public int getTimeout() {
+            return -1;
+        }
+
+    }
+
 }

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

@@ -0,0 +1,45 @@
+//
+// 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 com.cloud.hypervisor.kvm.resource.wrapper;
+
+import java.util.Map;
+
+import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.SetHostParamsCommand;
+import com.cloud.hypervisor.kvm.resource.LibvirtComputingResource;
+import com.cloud.resource.CommandWrapper;
+import com.cloud.resource.ResourceWrapper;
+
+@ResourceWrapper(handles =  SetHostParamsCommand.class)
+public final class LibvirtSetHostParamsCommandWrapper extends CommandWrapper<SetHostParamsCommand, Answer, LibvirtComputingResource> {
+
+    @Override
+    public Answer execute(final SetHostParamsCommand command, final LibvirtComputingResource libvirtComputingResource) {
+
+        final Map<String, String> params = command.getParams();
+        boolean success = libvirtComputingResource.getVirtRouterResource().configureHostParams(params);
+
+        if (!success) {
+            return new Answer(command, false, "Failed to set host parameters");
+        } else {
+            return new Answer(command, true, null);
+        }
+    }
+}