VR: don't drop aggregated commands when two jobs aggregate on one router

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

@@ -23,6 +23,7 @@
 import java.net.InetSocketAddress;
 import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -89,8 +90,12 @@ public class VirtualRoutingResource {
 
     protected Logger logger = LogManager.getLogger(getClass());
     private VirtualRouterDeployer _vrDeployer;
+    // Agent request threads run in parallel, so these maps are synchronized. The aggregation maps are
+    // only read and changed for a router while holding that router's lock from _vrLockMap.
     private Map<String, Queue<NetworkElementCommand>> _vrAggregateCommandsSet;
-    protected Map<String, Lock> _vrLockMap = new HashMap<String, Lock>();
+    // Router name -> failure details of the last aggregated apply on it, if that apply failed
+    private Map<String, String> _vrAggregateFailures;
+    protected Map<String, Lock> _vrLockMap = Collections.synchronizedMap(new HashMap<>());
 
     private String _name;
     private int _sleep;
@@ -107,13 +112,7 @@ public VirtualRoutingResource(VirtualRouterDeployer deployer) {
     public Answer executeRequest(final NetworkElementCommand cmd) {
         boolean aggregated = false;
         String routerName = cmd.getAccessDetail(NetworkElementCommand.ROUTER_NAME);
-        Lock lock;
-        if (_vrLockMap.containsKey(routerName)) {
-            lock = _vrLockMap.get(routerName);
-        } else {
-            lock = new ReentrantLock();
-            _vrLockMap.put(routerName, lock);
-        }
+        Lock lock = _vrLockMap.computeIfAbsent(routerName, k -> new ReentrantLock());
         lock.lock();
 
         try {
@@ -538,7 +537,8 @@ public boolean configure(final String name, final Map<String, Object> params) th
             throw new ConfigurationException("Unable to find the resource for VirtualRouterDeployer!");
         }
 
-        _vrAggregateCommandsSet = new HashMap<>();
+        _vrAggregateCommandsSet = Collections.synchronizedMap(new HashMap<>());
+        _vrAggregateFailures = Collections.synchronizedMap(new HashMap<>());
         return true;
     }
 
@@ -605,14 +605,31 @@ private Answer execute(AggregationControlCommand cmd) {
         assert cmd.getRouterAccessIp() != null;
 
         if (action == Action.Start) {
-            assert (!_vrAggregateCommandsSet.containsKey(routerName));
-
+            if (_vrAggregateCommandsSet.containsKey(routerName)) {
+                // Another job is aggregating commands for this router. Keep its queue: replacing it would
+                // drop the commands queued so far, and the first Finish applies the commands of both jobs.
+                logger.warn("Command aggregation is already in progress for router {}, joining it", routerName);
+                return new Answer(cmd, true, "Command aggregation already in progress");
+            }
             Queue<NetworkElementCommand> queue = new LinkedBlockingQueue<>();
             _vrAggregateCommandsSet.put(routerName, queue);
             return new Answer(cmd, true, "Command aggregation started");
         } else if (action == Action.Finish) {
             Queue<NetworkElementCommand> queue = _vrAggregateCommandsSet.get(routerName);
+            if (queue == null) {
+                // A concurrent job joined this aggregation and its Finish has already applied the queued
+                // commands; commands sent since then were applied directly. Report the result of that apply.
+                String failure = _vrAggregateFailures.get(routerName);
+                logger.warn("No command aggregation in progress for router {}, the last aggregated apply {}", routerName,
+                        failure == null ? "succeeded" : "failed: " + failure);
+                if (failure != null) {
+                    return new Answer(cmd, false, failure);
+                }
+                return new Answer(cmd, true, "Aggregated commands already applied");
+            }
             int answerCounts = 0;
+            boolean applied = false;
+            String failure = "Failed to apply aggregated commands";
             try {
                 StringBuilder sb = new StringBuilder();
                 sb.append("#Apache CloudStack Virtual Router Config File\n");
@@ -642,16 +659,24 @@ private Answer execute(AggregationControlCommand cmd) {
 
                 ExecutionResult result = applyConfigToVR(cmd.getRouterAccessIp(), fileConfigItem, timeout);
                 if (!result.isSuccess()) {
+                    failure = StringUtils.defaultIfBlank(result.getDetails(), failure);
                     return new Answer(cmd, false, result.getDetails());
                 }
 
                 result = applyConfigToVR(cmd.getRouterAccessIp(), scriptConfigItem, timeout);
                 if (!result.isSuccess()) {
+                    failure = StringUtils.defaultIfBlank(result.getDetails(), failure);
                     return new Answer(cmd, false, result.getDetails());
                 }
 
+                applied = true;
                 return new Answer(cmd, true, "Command aggregation finished");
             } finally {
+                if (applied) {
+                    _vrAggregateFailures.remove(routerName);
+                } else {
+                    _vrAggregateFailures.put(routerName, failure);
+                }
                 queue.clear();
                 _vrAggregateCommandsSet.remove(routerName);
             }

core/src/test/java/com/cloud/agent/resource/virtualnetwork/VirtualRoutingResourceAggregationTest.java

@@ -0,0 +1,155 @@
+// 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.resource.virtualnetwork;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.HashMap;
+import java.util.List;
+
+import org.joda.time.Duration;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.routing.AggregationControlCommand;
+import com.cloud.agent.api.routing.AggregationControlCommand.Action;
+import com.cloud.agent.api.routing.NetworkElementCommand;
+import com.cloud.agent.api.routing.SetPortForwardingRulesCommand;
+import com.cloud.agent.api.to.PortForwardingRuleTO;
+import com.cloud.utils.ExecutionResult;
+
+@RunWith(MockitoJUnitRunner.class)
+public class VirtualRoutingResourceAggregationTest {
+
+    private static final String ROUTER_IP = "169.254.3.4";
+    private static final String ROUTER_GUEST_IP = "10.200.1.1";
+    private static final String ROUTER_NAME = "r-4-VM";
+
+    @Mock
+    private VirtualRouterDeployer deployer;
+
+    private VirtualRoutingResource resource;
+
+    @Before
+    public void setup() throws Exception {
+        Mockito.when(deployer.prepareCommand(Mockito.any())).thenAnswer(invocation -> {
+            NetworkElementCommand cmd = invocation.getArgument(0);
+            cmd.setRouterAccessIp(ROUTER_IP);
+            return new ExecutionResult(true, null);
+        });
+        Mockito.lenient().when(deployer.cleanupCommand(Mockito.any())).thenReturn(new ExecutionResult(true, null));
+        Mockito.lenient().when(deployer.createFileInVR(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.anyString()))
+                .thenReturn(new ExecutionResult(true, null));
+        Mockito.lenient().when(deployer.executeInVR(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.any(Duration.class)))
+                .thenReturn(new ExecutionResult(true, null));
+        resource = new VirtualRoutingResource(deployer);
+        resource.configure("VRResource", new HashMap<>());
+    }
+
+    private Answer execute(NetworkElementCommand cmd) {
+        return resource.executeRequest(cmd);
+    }
+
+    private AggregationControlCommand control(Action action) {
+        return new AggregationControlCommand(action, ROUTER_NAME, ROUTER_IP, ROUTER_GUEST_IP);
+    }
+
+    private SetPortForwardingRulesCommand portForwarding(String publicIp) {
+        SetPortForwardingRulesCommand cmd = new SetPortForwardingRulesCommand(List.of(
+                new PortForwardingRuleTO(1, publicIp, 22, 22, "10.10.1.10", 22, 22, "TCP", false, false)));
+        cmd.setAccessDetail(NetworkElementCommand.ROUTER_NAME, ROUTER_NAME);
+        return cmd;
+    }
+
+    private List<String> appliedConfigFiles() {
+        ArgumentCaptor<String> content = ArgumentCaptor.forClass(String.class);
+        Mockito.verify(deployer, Mockito.atLeast(0)).createFileInVR(Mockito.eq(ROUTER_IP), Mockito.anyString(), Mockito.anyString(), content.capture());
+        return content.getAllValues();
+    }
+
+    @Test
+    public void overlappingAggregationsKeepCommandsOfBothJobs() {
+        assertTrue(execute(control(Action.Start)).getResult());
+        assertTrue(execute(portForwarding("64.1.1.10")).getResult());
+        assertTrue(execute(control(Action.Start)).getResult());
+        assertTrue(execute(portForwarding("64.1.1.20")).getResult());
+
+        assertTrue(execute(control(Action.Finish)).getResult());
+        List<String> files = appliedConfigFiles();
+        assertEquals(1, files.size());
+        assertTrue(files.get(0).contains("64.1.1.10"));
+        assertTrue(files.get(0).contains("64.1.1.20"));
+
+        Answer second = execute(control(Action.Finish));
+        assertTrue(second.getResult());
+        assertEquals(1, appliedConfigFiles().size());
+    }
+
+    @Test
+    public void finishOfJoinedAggregationReportsFailureOfTheApply() {
+        Mockito.when(deployer.executeInVR(Mockito.anyString(), Mockito.eq(VRScripts.VR_CFG), Mockito.anyString(), Mockito.any(Duration.class)))
+                .thenReturn(new ExecutionResult(false, "vr_cfg.sh failed"));
+
+        execute(control(Action.Start));
+        execute(portForwarding("64.1.1.10"));
+        execute(control(Action.Start));
+        execute(portForwarding("64.1.1.20"));
+
+        Answer first = execute(control(Action.Finish));
+        assertFalse(first.getResult());
+        Answer second = execute(control(Action.Finish));
+        assertFalse(second.getResult());
+        assertEquals("vr_cfg.sh failed", second.getDetails());
+    }
+
+    @Test
+    public void laterSuccessfulApplyClearsRecordedFailure() {
+        Mockito.when(deployer.executeInVR(Mockito.anyString(), Mockito.eq(VRScripts.VR_CFG), Mockito.anyString(), Mockito.any(Duration.class)))
+                .thenReturn(new ExecutionResult(false, "vr_cfg.sh failed"), new ExecutionResult(true, null));
+
+        execute(control(Action.Start));
+        assertFalse(execute(control(Action.Finish)).getResult());
+        execute(control(Action.Start));
+        assertTrue(execute(control(Action.Finish)).getResult());
+
+        assertTrue(execute(control(Action.Finish)).getResult());
+    }
+
+    @Test
+    public void finishWithoutStartDoesNotFail() {
+        Answer answer = execute(control(Action.Finish));
+        assertTrue(answer.getResult());
+        assertTrue(appliedConfigFiles().isEmpty());
+    }
+
+    @Test
+    public void commandsAfterFinishAreAppliedDirectly() {
+        execute(control(Action.Start));
+        execute(control(Action.Finish));
+
+        assertTrue(execute(portForwarding("64.1.1.30")).getResult());
+        Mockito.verify(deployer, Mockito.atLeastOnce()).executeInVR(Mockito.eq(ROUTER_IP), Mockito.eq(VRScripts.UPDATE_CONFIG), Mockito.anyString(), Mockito.any(Duration.class));
+    }
+}