From f7282a1d37527c2374a78af3200afbd97457e4e5 Mon Sep 17 00:00:00 2001 From: Brad House Date: Mon, 9 Nov 2026 19:07:49 +0000 Subject: [PATCH] VR: don't drop aggregated commands when two jobs aggregate on one router VirtualRoutingResource keeps one aggregation queue per router name. When two jobs aggregate commands for the same router at the same time, for example parallel tier deploys in one VPC: - the second Start replaces the first job's queue, so the commands the first job queued so far are never applied, while their answers already reported success; - the first Finish applies the second job's queue and removes it, so the second Finish finds no queue and fails with a NullPointerException on Queue.clear(). The per-router lock map and the aggregation map were also plain HashMaps changed from parallel agent request threads, and the lock was created with a check-then-put, so two threads could get different locks for the same router. - A Start while a queue exists now joins it instead of replacing it. The first Finish applies the commands of both jobs. - A Finish with no queue no longer fails with an NPE. Its commands were applied by the Finish that ran before it, or directly after that, so it returns the result of the last aggregated apply on the router: success, or that apply's failure details. - The maps are synchronized and the lock is created with computeIfAbsent. The aggregation of a single job behaves as before. Signed-off-by: Brad House --- .../VirtualRoutingResource.java | 47 ++++-- ...VirtualRoutingResourceAggregationTest.java | 155 ++++++++++++++++++ 2 files changed, 191 insertions(+), 11 deletions(-) create mode 100644 core/src/test/java/com/cloud/agent/resource/virtualnetwork/VirtualRoutingResourceAggregationTest.java diff --git a/core/src/main/java/com/cloud/agent/resource/virtualnetwork/VirtualRoutingResource.java b/core/src/main/java/com/cloud/agent/resource/virtualnetwork/VirtualRoutingResource.java index fb04193271ee..e281d89d99e0 100644 --- a/core/src/main/java/com/cloud/agent/resource/virtualnetwork/VirtualRoutingResource.java +++ b/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> _vrAggregateCommandsSet; - protected Map _vrLockMap = new HashMap(); + // Router name -> failure details of the last aggregated apply on it, if that apply failed + private Map _vrAggregateFailures; + protected Map _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 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 queue = new LinkedBlockingQueue<>(); _vrAggregateCommandsSet.put(routerName, queue); return new Answer(cmd, true, "Command aggregation started"); } else if (action == Action.Finish) { Queue 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); } diff --git a/core/src/test/java/com/cloud/agent/resource/virtualnetwork/VirtualRoutingResourceAggregationTest.java b/core/src/test/java/com/cloud/agent/resource/virtualnetwork/VirtualRoutingResourceAggregationTest.java new file mode 100644 index 000000000000..ca72de4bec70 --- /dev/null +++ b/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 appliedConfigFiles() { + ArgumentCaptor 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 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)); + } +}