From ce6e4d418d58d0e9e5c370aca84e1aa5e5592e9f Mon Sep 17 00:00:00 2001 From: Mitch <25337396+MitchDrage@users.noreply.github.com> Date: Fri, 6 Nov 2026 03:44:49 +0000 Subject: [PATCH 1/2] Allow exclusion of destination addresses from usage counters --- .../service/NetworkOrchestrationService.java | 51 ++++++++++++++++++ .../NetworkOrchestrationServiceTest.java | 51 ++++++++++++++++++ .../orchestration/NetworkOrchestrator.java | 2 +- .../ConfigurationManagerImpl.java | 26 +++++++++ .../VirtualNetworkApplianceManagerImpl.java | 6 +++ .../ConfigurationManagerImplTest.java | 25 +++++++++ systemvm/debian/opt/cloud/bin/cs/CsAddress.py | 24 +++++++-- systemvm/debian/opt/cloud/bin/cs/CsConfig.py | 24 +++++++++ systemvm/test/TestCsAddress.py | 53 ++++++++++++++++++- systemvm/test/TestCsConfig.py | 23 ++++++++ 10 files changed, 278 insertions(+), 7 deletions(-) create mode 100644 engine/api/src/test/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationServiceTest.java diff --git a/engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationService.java b/engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationService.java index 109a44488ec4..01dc9f0bab7a 100644 --- a/engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationService.java +++ b/engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationService.java @@ -16,6 +16,7 @@ // under the License. package org.apache.cloudstack.engine.orchestration.service; +import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -53,12 +54,14 @@ import com.cloud.user.User; import com.cloud.utils.fsm.NoTransitionException; import com.cloud.utils.Pair; +import com.cloud.utils.net.NetUtils; import com.cloud.vm.Nic; import com.cloud.vm.NicProfile; import com.cloud.vm.ReservationContext; import com.cloud.vm.VirtualMachine; import com.cloud.vm.VirtualMachine.Type; import com.cloud.vm.VirtualMachineProfile; +import com.googlecode.ipv6.IPv6Network; /** * NetworkManager manages the network for the different end users. @@ -138,6 +141,54 @@ public interface NetworkOrchestrationService { ConfigKey VmNetworkThrottlingRate = new ConfigKey("Network", Integer.class, "vm.network.throttling.rate", "200", "Default data transfer rate in megabits per second allowed in User vm's default network.", true, ConfigKey.Scope.Zone); + ConfigKey NetworkUsageExemptCidrs = new ConfigKey<>(String.class, "network.usage.exempt.cidrs", "Usage", "", + "Comma-separated IPv4 or IPv6 addresses or CIDRs. Traffic between guest networks and these addresses doesn't count " + + "towards virtual router network usage. Virtual routers only count IPv4 traffic, so they ignore IPv6 entries. " + + "Routers pick up a change the next time they start or restart.", + true, ConfigKey.Scope.Zone, null, null, null, null, null, ConfigKey.Kind.CSV, null); + + /** + * Tidies one entry of {@link #NetworkUsageExemptCidrs} into a CIDR, or returns null if it isn't a valid address or CIDR. + * For example, 192.0.2.5 becomes 192.0.2.5/32, 198.51.100.70/24 becomes 198.51.100.0/24 and 2001:db8::5 becomes + * 2001:db8::5/128. Callers refuse a /0 result, as it would exempt all traffic. + */ + static String normaliseUsageExemptCidr(String entry) { + if (NetUtils.isValidIp4(entry)) { + entry = entry + "/32"; + } else if (NetUtils.isValidIp6(entry)) { + entry = entry + "/128"; + } + if (NetUtils.isValidIp4Cidr(entry)) { + return NetUtils.getCidrSubNet(entry) + "/" + NetUtils.getCidr(entry).second(); + } + if (entry.contains(":") && NetUtils.isValidIp6Cidr(entry)) { + return IPv6Network.fromString(entry).toString(); + } + return null; + } + + /** + * Turns a {@link #NetworkUsageExemptCidrs} value into a tidy list with no repeats, using {@link #normaliseUsageExemptCidr}. + * Entries that aren't valid, and /0, are left out. Saving the setting already refuses them. + */ + static List parseUsageExemptCidrs(String value) { + final List cidrs = new ArrayList<>(); + if (value == null) { + return cidrs; + } + for (String entry : value.split(",")) { + entry = entry.trim(); + if (entry.isEmpty()) { + continue; + } + final String cidr = normaliseUsageExemptCidr(entry); + if (cidr != null && !cidr.endsWith("/0") && !cidrs.contains(cidr)) { + cidrs.add(cidr); + } + } + return cidrs; + } + List setupNetwork(Account owner, NetworkOffering offering, DeploymentPlan plan, String name, String displayText, boolean isDefault) throws ConcurrentOperationException; diff --git a/engine/api/src/test/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationServiceTest.java b/engine/api/src/test/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationServiceTest.java new file mode 100644 index 000000000000..40e834b98597 --- /dev/null +++ b/engine/api/src/test/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationServiceTest.java @@ -0,0 +1,51 @@ +/* + * 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.engine.orchestration.service; + +import java.util.List; + +import org.junit.Assert; +import org.junit.Test; + +public class NetworkOrchestrationServiceTest { + + @Test + public void parseUsageExemptCidrsTestNotSet() { + Assert.assertTrue(NetworkOrchestrationService.parseUsageExemptCidrs(null).isEmpty()); + Assert.assertTrue(NetworkOrchestrationService.parseUsageExemptCidrs(" ").isEmpty()); + } + + @Test + public void parseUsageExemptCidrsTestNormalisesEntries() { + List cidrs = NetworkOrchestrationService.parseUsageExemptCidrs("192.0.2.5, 198.51.100.70/24,,192.0.2.5/32"); + Assert.assertEquals(List.of("192.0.2.5/32", "198.51.100.0/24"), cidrs); + } + + @Test + public void parseUsageExemptCidrsTestNormalisesIpv6Entries() { + List cidrs = NetworkOrchestrationService.parseUsageExemptCidrs("2001:db8::5, 2001:db8:0:1::7/64,192.0.2.5"); + Assert.assertEquals(List.of("2001:db8::5/128", "2001:db8:0:1::/64", "192.0.2.5/32"), cidrs); + } + + @Test + public void parseUsageExemptCidrsTestSkipsInvalidEntries() { + List cidrs = NetworkOrchestrationService.parseUsageExemptCidrs("not-an-ip,0.0.0.0/0,::/0,10.0.0.0/33,192.168.1.300,172.16.0.0/12"); + Assert.assertEquals(List.of("172.16.0.0/12"), cidrs); + } +} diff --git a/engine/orchestration/src/main/java/org/apache/cloudstack/engine/orchestration/NetworkOrchestrator.java b/engine/orchestration/src/main/java/org/apache/cloudstack/engine/orchestration/NetworkOrchestrator.java index c1697819174d..df44c035c18c 100644 --- a/engine/orchestration/src/main/java/org/apache/cloudstack/engine/orchestration/NetworkOrchestrator.java +++ b/engine/orchestration/src/main/java/org/apache/cloudstack/engine/orchestration/NetworkOrchestrator.java @@ -5117,6 +5117,6 @@ public ConfigKey[] getConfigKeys() { GuestDomainSuffix, NetworkThrottlingRate, VmNetworkThrottlingRate, MinVRVersion, DhcpLeaseTimeout, PromiscuousMode, MacAddressChanges, ForgedTransmits, MacLearning, RollingRestartEnabled, TUNGSTEN_ENABLED, NSX_ENABLED, NETRIS_ENABLED, NETWORK_LB_HAPROXY_MAX_CONN, - NETWORK_LB_HAPROXY_IDLE_TIMEOUT}; + NETWORK_LB_HAPROXY_IDLE_TIMEOUT, NetworkUsageExemptCidrs}; } } diff --git a/server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java b/server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java index 17dbced54233..6b2af820fc2e 100644 --- a/server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java +++ b/server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java @@ -1561,6 +1561,12 @@ protected String validateValueRange(String name, String value, Class type, Co } } } + if (NetworkOrchestrationService.NetworkUsageExemptCidrs.key().equalsIgnoreCase(name) && StringUtils.isNotBlank(value)) { + final String error = validateUsageExemptCidrs(name, value); + if (error != null) { + return error; + } + } } validateIpAddressRelatedConfigValues(name, value); @@ -1576,6 +1582,26 @@ protected String validateValueRange(String name, String value, Class type, Co return validateIfStringValueIsInRange(name, value, range); } + /** + * Checks that each entry is an IPv4 or IPv6 address or CIDR. We refuse /0 because it would exempt all traffic. + */ + protected String validateUsageExemptCidrs(String name, String value) { + for (String entry : value.split(",")) { + entry = entry.trim(); + if (entry.isEmpty()) { + continue; + } + final String cidr = NetworkOrchestrationService.normaliseUsageExemptCidr(entry); + if (cidr == null) { + return String.format("[%s] is not a valid IP address or CIDR for %s.", entry, name); + } + if (cidr.endsWith("/0")) { + return String.format("[%s] would exempt all traffic, so it can't be used for %s.", entry, name); + } + } + return null; + } + /** * Validates configuration values for the given name, value, and type. *
    diff --git a/server/src/main/java/com/cloud/network/router/VirtualNetworkApplianceManagerImpl.java b/server/src/main/java/com/cloud/network/router/VirtualNetworkApplianceManagerImpl.java index fe167b7e7e76..1a9539757a2d 100644 --- a/server/src/main/java/com/cloud/network/router/VirtualNetworkApplianceManagerImpl.java +++ b/server/src/main/java/com/cloud/network/router/VirtualNetworkApplianceManagerImpl.java @@ -2089,6 +2089,12 @@ public boolean finalizeVirtualMachineProfile(final VirtualMachineProfile profile buf.append(" exposedns=true"); } + final List usageExemptCidrs = NetworkOrchestrationService.parseUsageExemptCidrs( + NetworkOrchestrationService.NetworkUsageExemptCidrs.valueIn(dc.getId())); + if (!usageExemptCidrs.isEmpty()) { + buf.append(" usage_exempt_cidrs=").append(String.join(",", usageExemptCidrs)); + } + if (Boolean.parseBoolean(_configDao.getValue(Config.BaremetalProvisionDoneNotificationEnabled.key()))) { final QueryBuilder acntq = QueryBuilder.create(UserVO.class); acntq.and(acntq.entity().getUsername(), SearchCriteria.Op.EQ, "baremetal-system-account"); diff --git a/server/src/test/java/com/cloud/configuration/ConfigurationManagerImplTest.java b/server/src/test/java/com/cloud/configuration/ConfigurationManagerImplTest.java index 9a0b150780e4..a32ef1cdad72 100644 --- a/server/src/test/java/com/cloud/configuration/ConfigurationManagerImplTest.java +++ b/server/src/test/java/com/cloud/configuration/ConfigurationManagerImplTest.java @@ -66,6 +66,7 @@ import org.apache.cloudstack.api.command.admin.zone.DeleteZoneCmd; import org.apache.cloudstack.config.Configuration; import org.apache.cloudstack.context.CallContext; +import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService; import org.apache.cloudstack.engine.subsystem.api.storage.ZoneScope; import org.apache.cloudstack.framework.config.ConfigDepot; import org.apache.cloudstack.framework.config.ConfigKey; @@ -861,6 +862,30 @@ public void validateValueRangeTestReturnsNotNullWhenConfigKeyHasRangeAndValueIsI Assert.assertNotNull(configurationManagerImplSpy.validateValueRange(NetworkModel.MACIdentifier.key(), "-1", Integer.class, null)); } + private String validateUsageExemptCidrs(String value) { + return configurationManagerImplSpy.validateValueRange(NetworkOrchestrationService.NetworkUsageExemptCidrs.key(), value, String.class, null); + } + + @Test + public void validateValueRangeTestAcceptsUsageExemptAddressesAndCidrs() { + Assert.assertNull(validateUsageExemptCidrs("")); + Assert.assertNull(validateUsageExemptCidrs("192.0.2.5, 198.51.100.70/24,,203.0.113.0/24")); + Assert.assertNull(validateUsageExemptCidrs("2001:db8::5,2001:db8:0:1::7/64")); + } + + @Test + public void validateValueRangeTestRejectsInvalidUsageExemptEntry() { + Assert.assertNotNull(validateUsageExemptCidrs("192.0.2.5,bogus")); + Assert.assertNotNull(validateUsageExemptCidrs("192.0.2.300")); + Assert.assertNotNull(validateUsageExemptCidrs("2001:db8::/129")); + } + + @Test + public void validateValueRangeTestRejectsUsageExemptMatchAll() { + Assert.assertNotNull(validateUsageExemptCidrs("0.0.0.0/0")); + Assert.assertNotNull(validateUsageExemptCidrs("::/0")); + } + @Test public void validateValueRangeTestValidatesValueWhenConfigHasRange() { Config config = Config.SecStorageEncryptCopy; diff --git a/systemvm/debian/opt/cloud/bin/cs/CsAddress.py b/systemvm/debian/opt/cloud/bin/cs/CsAddress.py index 08aee14f43bc..01f8459ecb56 100755 --- a/systemvm/debian/opt/cloud/bin/cs/CsAddress.py +++ b/systemvm/debian/opt/cloud/bin/cs/CsAddress.py @@ -473,6 +473,22 @@ def setup_router_control(self): self.fw.append(["filter", "", "-P INPUT DROP"]) self.fw.append(["filter", "", "-P FORWARD DROP"]) + def add_usage_stats_jump(self, stats_chain): + """ Sends forwarded traffic to the usage counters. If exempt CIDRs are set, traffic passes + through _EXEMPT first, which skips the counters for those addresses. + We keep the usage chain exactly as it was, because the agent reads its counters by position. """ + exempt_cidrs = self.config.get_usage_exempt_cidrs() + if not exempt_cidrs: + self.fw.append(["", "front", "-A FORWARD -j %s" % stats_chain]) + return + exempt_chain = "%s_EXEMPT" % stats_chain + self.fw.append(["", "front", "-A FORWARD -j %s" % exempt_chain]) + self.fw.append(["", "", "-A %s -j %s" % (exempt_chain, stats_chain)]) + for cidr in exempt_cidrs: + # insert at the top, so a CIDR added later still comes before the jump to the counters + self.fw.append(["", "front", "-A %s -d %s -j RETURN" % (exempt_chain, cidr)]) + self.fw.append(["", "front", "-A %s -s %s -j RETURN" % (exempt_chain, cidr)]) + def fw_router(self): if self.config.is_vpc() or self.config.is_routed() or self.config.is_dhcp(): return @@ -501,8 +517,7 @@ def fw_router(self): "-I VPN_%s -m state --state RELATED,ESTABLISHED -j ACCEPT" % self.address['public_ip']]) self.fw.append(["mangle", "", "-A VPN_%s -j RETURN" % self.address['public_ip']]) - self.fw.append( - ["", "front", "-A FORWARD -j NETWORK_STATS_%s" % self.dev]) + self.add_usage_stats_jump("NETWORK_STATS_%s" % self.dev) self.fw.append( ["", "front", "-A INPUT -j NETWORK_STATS_%s" % self.dev]) self.fw.append( @@ -553,7 +568,7 @@ def fw_router(self): self.fw.append( ["filter", "", "-A FORWARD -i eth0 -o eth0 -m state --state RELATED,ESTABLISHED -j ACCEPT"]) - self.fw.append(['', 'front', '-A FORWARD -j NETWORK_STATS']) + self.add_usage_stats_jump("NETWORK_STATS") self.fw.append(['', 'front', '-A INPUT -j NETWORK_STATS']) self.fw.append(['', 'front', '-A OUTPUT -j NETWORK_STATS']) self.fw.append(['', '', '-A NETWORK_STATS -i eth0 -o eth2']) @@ -688,8 +703,7 @@ def fw_vpcrouter(self): ["mangle", "", "-A VPN_STATS_%s -o %s -m mark --mark 0x525/0xffffffff" % (self.dev, self.dev)]) self.fw.append( ["mangle", "", "-A VPN_STATS_%s -i %s -m mark --mark 0x524/0xffffffff" % (self.dev, self.dev)]) - self.fw.append( - ["", "front", "-A FORWARD -j NETWORK_STATS_%s" % self.dev]) + self.add_usage_stats_jump("NETWORK_STATS_%s" % self.dev) self.fw.append( ["", "front", "-A INPUT -j NETWORK_STATS_%s" % self.dev]) self.fw.append( diff --git a/systemvm/debian/opt/cloud/bin/cs/CsConfig.py b/systemvm/debian/opt/cloud/bin/cs/CsConfig.py index 5218e8111161..e31421f85ac2 100755 --- a/systemvm/debian/opt/cloud/bin/cs/CsConfig.py +++ b/systemvm/debian/opt/cloud/bin/cs/CsConfig.py @@ -18,6 +18,7 @@ from .CsDatabag import CsCmdLine, CsGuestNetwork from .CsAddress import CsAddress +import ipaddress import logging @@ -114,6 +115,29 @@ def use_extdns(self): def expose_dns(self): return self.cmdline().idata().get('exposedns', 'false') == 'true' + def get_usage_exempt_cidrs(self): + """ The IPv4 CIDRs to leave out of network usage, tidied to match how iptables-save prints them. + Without that, each run would see the rules as changed and rewrite them. + IPv6 entries are skipped, as the router only counts IPv4 traffic. """ + cidrs = [] + for value in self.cmdline().idata().get('usage_exempt_cidrs', '').split(','): + value = value.strip() + if not value: + continue + try: + network = ipaddress.ip_network(value, strict=False) + except ValueError: + logging.warning("Skipping %s in usage exempt CIDRs: not a valid IP address or CIDR", value) + continue + if network.version != 4: + continue + if network.prefixlen == 0: + logging.warning("Skipping %s in usage exempt CIDRs: it would exempt all traffic", value) + continue + if str(network) not in cidrs: + cidrs.append(str(network)) + return cidrs + def use_router_ip_as_resolver(self): return self.cl.get_use_router_ip_as_resolver() diff --git a/systemvm/test/TestCsAddress.py b/systemvm/test/TestCsAddress.py index 695cc6d7ba4b..20578859583c 100644 --- a/systemvm/test/TestCsAddress.py +++ b/systemvm/test/TestCsAddress.py @@ -27,6 +27,8 @@ def __init__(self): self.fw = [] self.nft_ipv4_fw = [] self.nft_ipv4_acl = [] + self.usage_exempt_cidrs = [] + self.vpc = True def get_fw(self): return self.fw @@ -43,14 +45,20 @@ def cmdline(self): return cl def is_vpc(self): - return True + return self.vpc def is_routed(self): return False + def is_dhcp(self): + return False + def has_public_network(self): return True + def get_usage_exempt_cidrs(self): + return self.usage_exempt_cidrs + class TestCsAddress(unittest.TestCase): @@ -113,6 +121,49 @@ def test_acl_outbound_ends_with_return_on_private_gateway(self): "gateway": "172.16.0.1", "public_ip": "172.16.0.10", "source_nat": False}) self.assertEqual(rules, [["mangle", "", "-A ACL_OUTBOUND_eth3 -j RETURN"]]) + def usage_rules(self, exempt_cidrs, vpc=True): + config = FakeConfig() + config.usage_exempt_cidrs = exempt_cidrs + config.vpc = vpc + with mock.patch.object(CsIP, "list"): + ip = CsIP("eth3", config) + ip.setAddress({"nw_type": "public", "public_ip": "203.0.113.10", "network": "203.0.113.0/24", + "gateway": "203.0.113.1", "source_nat": True}) + with mock.patch("cs.CsAddress.CsStaticRoutes"): + if vpc: + ip.fw_vpcrouter() + else: + ip.fw_router() + return [fw for fw in config.fw if fw[0] == "" and "NETWORK_STATS" in fw[2]] + + def test_usage_jump_unchanged_without_exempt_cidrs(self): + rules = self.usage_rules([]) + self.assertIn(["", "front", "-A FORWARD -j NETWORK_STATS_eth3"], rules) + self.assertFalse(any("_EXEMPT" in fw[2] for fw in rules)) + + def test_vpc_usage_jump_goes_through_exempt_chain(self): + rules = self.usage_rules(["192.0.2.5/32"]) + self.assertNotIn(["", "front", "-A FORWARD -j NETWORK_STATS_eth3"], rules) + exempt = [fw for fw in rules if "NETWORK_STATS_eth3_EXEMPT" in fw[2]] + self.assertEqual(exempt, [ + ["", "front", "-A FORWARD -j NETWORK_STATS_eth3_EXEMPT"], + ["", "", "-A NETWORK_STATS_eth3_EXEMPT -j NETWORK_STATS_eth3"], + ["", "front", "-A NETWORK_STATS_eth3_EXEMPT -d 192.0.2.5/32 -j RETURN"], + ["", "front", "-A NETWORK_STATS_eth3_EXEMPT -s 192.0.2.5/32 -j RETURN"], + ]) + + def test_usage_counting_chain_unchanged_with_exempt_cidrs(self): + # the agent reads these counters by position, so the usage chain has to stay exactly the same + counting = lambda rules: [fw for fw in rules if fw[2].startswith("-A NETWORK_STATS_eth3 ")] + self.assertEqual(counting(self.usage_rules(["192.0.2.5/32"])), counting(self.usage_rules([]))) + + def test_isolated_usage_jumps_go_through_exempt_chains(self): + rules = self.usage_rules(["192.0.2.5/32"], vpc=False) + forward = [fw[2] for fw in rules if fw[2].startswith("-A FORWARD")] + self.assertEqual(sorted(forward), ["-A FORWARD -j NETWORK_STATS_EXEMPT", "-A FORWARD -j NETWORK_STATS_eth3_EXEMPT"]) + self.assertIn(["", "", "-A NETWORK_STATS_EXEMPT -j NETWORK_STATS"], rules) + self.assertIn(["", "front", "-A NETWORK_STATS_EXEMPT -d 192.0.2.5/32 -j RETURN"], rules) + if __name__ == '__main__': unittest.main() diff --git a/systemvm/test/TestCsConfig.py b/systemvm/test/TestCsConfig.py index 02f65a2c99f4..739610347292 100644 --- a/systemvm/test/TestCsConfig.py +++ b/systemvm/test/TestCsConfig.py @@ -16,6 +16,7 @@ # under the License. import unittest +from unittest import mock from cs.CsConfig import CsConfig import merge @@ -29,6 +30,28 @@ def test_ini(self): csconfig = CsConfig() self.assertTrue(csconfig is not None) + def usage_exempt_cidrs(self, cmdline): + cl = mock.Mock() + cl.idata.return_value = cmdline + with mock.patch.object(CsConfig, "cmdline", return_value=cl): + return CsConfig().get_usage_exempt_cidrs() + + def test_usage_exempt_cidrs_not_set(self): + self.assertEqual(self.usage_exempt_cidrs({}), []) + + def test_usage_exempt_cidrs_are_written_as_iptables_prints_them(self): + cidrs = self.usage_exempt_cidrs({"usage_exempt_cidrs": "192.0.2.5/32, 198.51.100.70/24,192.0.2.5"}) + self.assertEqual(cidrs, ["192.0.2.5/32", "198.51.100.0/24"]) + + def test_usage_exempt_cidrs_skip_invalid_and_match_all(self): + cidrs = self.usage_exempt_cidrs({"usage_exempt_cidrs": "not-an-ip,0.0.0.0/0,192.168.1.300,172.16.0.0/12"}) + self.assertEqual(cidrs, ["172.16.0.0/12"]) + + def test_usage_exempt_cidrs_skip_ipv6(self): + # the router only counts IPv4 traffic, so IPv6 entries have nothing to exempt + cidrs = self.usage_exempt_cidrs({"usage_exempt_cidrs": "2001:db8::5/128,192.0.2.5/32,2001:db8:0:1::/64"}) + self.assertEqual(cidrs, ["192.0.2.5/32"]) + if __name__ == '__main__': unittest.main() From 3d943aad048889f50e62a56f71cbfa1b9a16fef2 Mon Sep 17 00:00:00 2001 From: Mitch <25337396+MitchDrage@users.noreply.github.com> Date: Sat, 7 Nov 2026 10:43:02 +0000 Subject: [PATCH 2/2] Allow exclusion of destination addresses from usage on KVM shared networks --- debian/control | 2 +- .../service/NetworkOrchestrationService.java | 7 +- .../java/com/cloud/agent/AgentManager.java | 6 + .../cloud/agent/manager/AgentManagerImpl.java | 27 + .../agent/manager/AgentManagerImplTest.java | 33 ++ packaging/el8/cloud.spec | 1 + packaging/suse15/cloud.spec | 1 + .../resource/LibvirtComputingResource.java | 51 +- .../kvm/resource/UsageExemptBilling.java | 195 +++++++ .../kvm/resource/UsageExemptCounters.java | 478 ++++++++++++++++++ ...ibvirtGetVmNetworkStatsCommandWrapper.java | 24 +- .../kvm/resource/UsageExemptBillingTest.java | 131 +++++ .../kvm/resource/UsageExemptCountersTest.java | 370 ++++++++++++++ .../ConfigurationManagerImpl.java | 7 + 14 files changed, 1316 insertions(+), 17 deletions(-) create mode 100644 plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/UsageExemptBilling.java create mode 100644 plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/UsageExemptCounters.java create mode 100644 plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/UsageExemptBillingTest.java create mode 100644 plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/UsageExemptCountersTest.java diff --git a/debian/control b/debian/control index 30504d4c981b..8b4af70f7b5c 100644 --- a/debian/control +++ b/debian/control @@ -24,7 +24,7 @@ Description: CloudStack server library Package: cloudstack-agent Architecture: all -Depends: ${python:Depends}, ${python3:Depends}, openjdk-17-jre-headless | java17-runtime-headless | java17-runtime | zulu-17 | openjdk-21-jre-headless | java21-runtime-headless | java21-runtime | zulu-21, cloudstack-common (= ${source:Version}), lsb-base (>= 9), openssh-client, qemu-kvm (>= 2.5) | qemu-system-x86 (>= 5.2), libvirt-bin (>= 1.3) | libvirt-daemon-system (>= 3.0), iproute2, ebtables, vlan, ipset, python3-libvirt, ethtool, iptables, cryptsetup, rng-tools, rsync, ovmf, swtpm, lsb-release, ufw, apparmor, cpu-checker, libvirt-daemon-driver-storage-rbd, sysstat, python3-libnbd, socat +Depends: ${python:Depends}, ${python3:Depends}, openjdk-17-jre-headless | java17-runtime-headless | java17-runtime | zulu-17 | openjdk-21-jre-headless | java21-runtime-headless | java21-runtime | zulu-21, cloudstack-common (= ${source:Version}), lsb-base (>= 9), openssh-client, qemu-kvm (>= 2.5) | qemu-system-x86 (>= 5.2), libvirt-bin (>= 1.3) | libvirt-daemon-system (>= 3.0), iproute2, ebtables, vlan, ipset, nftables, python3-libvirt, ethtool, iptables, cryptsetup, rng-tools, rsync, ovmf, swtpm, lsb-release, ufw, apparmor, cpu-checker, libvirt-daemon-driver-storage-rbd, sysstat, python3-libnbd, socat Recommends: init-system-helpers Conflicts: cloud-agent, cloud-agent-libs, cloud-agent-deps, cloud-agent-scripts Description: CloudStack agent diff --git a/engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationService.java b/engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationService.java index 01dc9f0bab7a..c8bcd7882298 100644 --- a/engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationService.java +++ b/engine/api/src/main/java/org/apache/cloudstack/engine/orchestration/service/NetworkOrchestrationService.java @@ -142,9 +142,10 @@ public interface NetworkOrchestrationService { "Default data transfer rate in megabits per second allowed in User vm's default network.", true, ConfigKey.Scope.Zone); ConfigKey NetworkUsageExemptCidrs = new ConfigKey<>(String.class, "network.usage.exempt.cidrs", "Usage", "", - "Comma-separated IPv4 or IPv6 addresses or CIDRs. Traffic between guest networks and these addresses doesn't count " + - "towards virtual router network usage. Virtual routers only count IPv4 traffic, so they ignore IPv6 entries. " + - "Routers pick up a change the next time they start or restart.", + "Comma-separated IPv4 or IPv6 addresses or CIDRs. Traffic between guest VMs and these addresses doesn't count " + + "towards network usage. Virtual routers only count IPv4 traffic, so they ignore IPv6 entries, and pick up a change " + + "the next time they start or restart. On shared networks, KVM hosts exempt both IPv4 and IPv6 and pick up a change " + + "straight away.", true, ConfigKey.Scope.Zone, null, null, null, null, null, ConfigKey.Kind.CSV, null); /** diff --git a/engine/components-api/src/main/java/com/cloud/agent/AgentManager.java b/engine/components-api/src/main/java/com/cloud/agent/AgentManager.java index 4d63fae33560..58441c6a259a 100644 --- a/engine/components-api/src/main/java/com/cloud/agent/AgentManager.java +++ b/engine/components-api/src/main/java/com/cloud/agent/AgentManager.java @@ -175,6 +175,12 @@ enum TapAgentsAction { void propagateChangeToAgents(Map params); + /** + * Sends network.usage.exempt.cidrs to the KVM hosts in one zone, or in every zone when zoneId is null, + * using the value for each host's zone. + */ + void propagateUsageExemptCidrsToAgents(Long zoneId); + boolean transferDirectAgentsFromMS(String fromMsUuid, long fromMsId, long timeoutDurationInMs, boolean excludeHostsInMaintenance); int getHostSshPort(HostVO host); diff --git a/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java b/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java index 9eab15214aeb..fe389af366bc 100644 --- a/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java +++ b/engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java @@ -23,6 +23,7 @@ import java.nio.channels.ClosedChannelException; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.List; @@ -2275,6 +2276,7 @@ public void processConnect(final Host host, final StartupCommand cmd, final bool params.put(Config.MigrateWait.toString(), _configDao.getValue(Config.MigrateWait.toString())); params.put(NetworkOrchestrationService.TUNGSTEN_ENABLED.key(), String.valueOf(NetworkOrchestrationService.TUNGSTEN_ENABLED.valueIn(host.getDataCenterId()))); params.put(ReconcileCommandService.ReconcileCommandsEnabled.key(), String.valueOf(_reconcileCommandsEnabled)); + params.put(NetworkOrchestrationService.NetworkUsageExemptCidrs.key(), getUsageExemptCidrsParam(host.getDataCenterId())); try { SetHostParamsCommand cmds = new SetHostParamsCommand(params); @@ -2353,6 +2355,31 @@ public void propagateChangeToAgents(Map params) { } } + @Override + public void propagateUsageExemptCidrsToAgents(Long zoneId) { + final Map> hostsPerZone = getHostsPerZone(); + if (hostsPerZone == null) { + return; + } + logger.debug("Sending {} to the agents", NetworkOrchestrationService.NetworkUsageExemptCidrs.key()); + for (Map.Entry> zone : hostsPerZone.entrySet()) { + if (zoneId != null && !zoneId.equals(zone.getKey())) { + continue; + } + final Map params = new HashMap<>(); + params.put(NetworkOrchestrationService.NetworkUsageExemptCidrs.key(), getUsageExemptCidrsParam(zone.getKey())); + sendCommandToAgents(Collections.singletonMap(zone.getKey(), zone.getValue()), params); + } + } + + /** + * The zone's exempt list, tidied and joined with commas. An empty string tells the host the list is empty. + */ + protected String getUsageExemptCidrsParam(long zoneId) { + return String.join(",", NetworkOrchestrationService.parseUsageExemptCidrs( + NetworkOrchestrationService.NetworkUsageExemptCidrs.valueIn(zoneId))); + } + @Override public boolean transferDirectAgentsFromMS(String fromMsUuid, long fromMsId, long timeoutDurationInMs, boolean excludeHostsInMaintenance) { return true; diff --git a/engine/orchestration/src/test/java/com/cloud/agent/manager/AgentManagerImplTest.java b/engine/orchestration/src/test/java/com/cloud/agent/manager/AgentManagerImplTest.java index 2377bbaefbc8..9d08f6264cd0 100644 --- a/engine/orchestration/src/test/java/com/cloud/agent/manager/AgentManagerImplTest.java +++ b/engine/orchestration/src/test/java/com/cloud/agent/manager/AgentManagerImplTest.java @@ -20,6 +20,7 @@ import com.cloud.agent.api.Answer; import com.cloud.agent.api.ReadyAnswer; import com.cloud.agent.api.ReadyCommand; +import com.cloud.agent.api.SetHostParamsCommand; import com.cloud.agent.api.StartupCommand; import com.cloud.agent.api.StartupRoutingCommand; import com.cloud.exception.ConnectionException; @@ -37,7 +38,10 @@ import org.mockito.Mockito; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; public class AgentManagerImplTest { @@ -172,4 +176,33 @@ public void testGetHostSshPortWithKVMHostCustomPort() { int hostSshPort = mgr.getHostSshPort(host); Assert.assertEquals(3922, hostSshPort); } + + private void mockUsageExemptPush() { + Map> hostsPerZone = new HashMap<>(); + hostsPerZone.put(1L, Arrays.asList(11L, 12L)); + hostsPerZone.put(2L, Collections.singletonList(21L)); + Mockito.doReturn(hostsPerZone).when(mgr).getHostsPerZone(); + Mockito.doReturn("192.0.2.5/32").when(mgr).getUsageExemptCidrsParam(Mockito.anyLong()); + Answer answer = Mockito.mock(Answer.class); + Mockito.when(answer.getResult()).thenReturn(true); + Mockito.doReturn(answer).when(mgr).easySend(Mockito.anyLong(), Mockito.any(SetHostParamsCommand.class)); + } + + @Test + public void propagateUsageExemptCidrsToAgentsTestOneZone() { + mockUsageExemptPush(); + mgr.propagateUsageExemptCidrsToAgents(1L); + Mockito.verify(mgr).easySend(Mockito.eq(11L), Mockito.any(SetHostParamsCommand.class)); + Mockito.verify(mgr).easySend(Mockito.eq(12L), Mockito.any(SetHostParamsCommand.class)); + Mockito.verify(mgr, Mockito.never()).easySend(Mockito.eq(21L), Mockito.any(SetHostParamsCommand.class)); + } + + @Test + public void propagateUsageExemptCidrsToAgentsTestAllZones() { + mockUsageExemptPush(); + mgr.propagateUsageExemptCidrsToAgents(null); + Mockito.verify(mgr, Mockito.times(3)).easySend(Mockito.anyLong(), Mockito.any(SetHostParamsCommand.class)); + Mockito.verify(mgr).getUsageExemptCidrsParam(1L); + Mockito.verify(mgr).getUsageExemptCidrsParam(2L); + } } diff --git a/packaging/el8/cloud.spec b/packaging/el8/cloud.spec index 9d8cab846164..c5c6c5375030 100644 --- a/packaging/el8/cloud.spec +++ b/packaging/el8/cloud.spec @@ -114,6 +114,7 @@ Requires: ethtool Requires: (net-tools or net-tools-deprecated) Requires: iproute Requires: ipset +Requires: nftables Requires: perl Requires: rsync Requires: cifs-utils diff --git a/packaging/suse15/cloud.spec b/packaging/suse15/cloud.spec index 04ab66904287..dafb4e3d411f 100644 --- a/packaging/suse15/cloud.spec +++ b/packaging/suse15/cloud.spec @@ -114,6 +114,7 @@ Requires: ethtool Requires: (net-tools or net-tools-deprecated) Requires: iproute Requires: ipset +Requires: nftables Requires: perl Requires: rsync Requires: cifs-utils diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java index acffa395494d..8b486c58390a 100644 --- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java +++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java @@ -613,6 +613,7 @@ public class LibvirtComputingResource extends ServerResourceBase implements Serv protected MemStat memStat = new MemStat(dom0MinMem, dom0OvercommitMem); private final LibvirtUtilitiesHelper libvirtUtilitiesHelper = new LibvirtUtilitiesHelper(); private LibvirtDomainListener libvirtDomainListener; + private final UsageExemptCounters usageExemptCounters = new UsageExemptCounters(); protected Boolean enableManuallySettingCpuTopologyOnKvmVm = AgentPropertiesFileHandler.getPropertyValue(AgentProperties.ENABLE_MANUALLY_SETTING_CPU_TOPOLOGY_ON_KVM_VM); @@ -1780,10 +1781,19 @@ public boolean configureHostParams(final Map params) { isTungstenEnabled = Boolean.parseBoolean(params.get(NetworkOrchestrationService.TUNGSTEN_ENABLED.key())); } - if (params.get(ReconcileCommandService.ReconcileCommandsEnabled.key()) != null) { - isReconcileCommandsEnabled = Boolean.parseBoolean(params.get(ReconcileCommandService.ReconcileCommandsEnabled.key())); + final String usageExemptCidrs = params.get(NetworkOrchestrationService.NetworkUsageExemptCidrs.key()); + if (usageExemptCidrs != null && hypervisorType == HypervisorType.KVM) { + usageExemptCounters.setTapOwners(this::getTapOwners); + usageExemptCounters.setCidrs(NetworkOrchestrationService.parseUsageExemptCidrs(usageExemptCidrs)); } - if (isReconcileCommandsEnabled) { + + final String reconcileCommandsEnabled = params.get(ReconcileCommandService.ReconcileCommandsEnabled.key()); + if (reconcileCommandsEnabled != null) { + isReconcileCommandsEnabled = Boolean.parseBoolean(reconcileCommandsEnabled); + } + // Only on connect, which is the only time the server sends this setting. Settings pushed while the agent is + // running must leave the logs of commands still in progress alone. + if (isReconcileCommandsEnabled && reconcileCommandsEnabled != null) { File commandsLogPath = new File(COMMANDS_LOG_PATH); if (!commandsLogPath.exists()) { commandsLogPath.mkdirs(); @@ -5180,6 +5190,38 @@ private int executeBashScriptAndRetrieveExitValue(final String script) { return command.getExitValue(); } + public UsageExemptCounters getUsageExemptCounters() { + return usageExemptCounters; + } + + /** + * Maps each running VM's NIC device name to the VM's name, so the usage exemption counters can skip system VMs. + * Returns null if libvirt can't be asked. + */ + protected Map getTapOwners() { + try { + final Connect conn = LibvirtConnection.getConnection(); + final Map owners = new HashMap<>(); + for (final int id : conn.listDomains()) { + final Domain dm = conn.domainLookupByID(id); + try { + final String vmName = dm.getName(); + for (final InterfaceDef nic : getInterfaces(conn, vmName)) { + if (nic.getDevName() != null) { + owners.put(nic.getDevName(), vmName); + } + } + } finally { + dm.free(); + } + } + return owners; + } catch (final LibvirtException e) { + LOGGER.debug("Failed to list the running VMs' NICs for the usage exemption counters", e); + return null; + } + } + public List getVmNetworkStat(Connect conn, String vmName) throws LibvirtException { Domain dm = null; try { @@ -5192,7 +5234,8 @@ public List getVmNetworkStat(Connect conn, String vmName) t for (InterfaceDef nic : nics) { DomainInterfaceStats nicStats = dm.interfaceStats(nic.getDevName()); String macAddress = nic.getMacAddress(); - VmNetworkStatsEntry stat = new VmNetworkStatsEntry(vmName, macAddress, nicStats.tx_bytes, nicStats.rx_bytes); + long[] usage = usageExemptCounters.subtractExempt(nic.getDevName(), nicStats.tx_bytes, nicStats.rx_bytes); + VmNetworkStatsEntry stat = new VmNetworkStatsEntry(vmName, macAddress, usage[0], usage[1]); stats.add(stat); } diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/UsageExemptBilling.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/UsageExemptBilling.java new file mode 100644 index 000000000000..2c11f6147667 --- /dev/null +++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/UsageExemptBilling.java @@ -0,0 +1,195 @@ +// 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; + +import java.io.IOException; +import java.lang.reflect.Type; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.HashMap; +import java.util.Map; +import java.util.function.BiPredicate; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; + +/** + * The billing side of usage exemption on KVM hosts. It takes each tap's exempt bytes off its libvirt totals and keeps the + * reported numbers safe to bill. {@link UsageExemptCounters} reads the exempt bytes from nftables and passes them in. + *
      + *
    • The management server reads a smaller number as a counter reset and bills the previous one again. So for each tap + * the reported numbers never go down. The record of what was reported lives in /run. It survives an agent restart, and + * a host reboot clears it along with the tap counters.
      • + *
    • When a tap's exempt counters are wiped, for example by someone flushing nftables, the bytes they had counted are + * kept and still taken off.
      • + *
    • A tap with a new ifindex, or with libvirt totals that went down, is a new tap. Its record starts again from zero, + * the same as libvirt's counters do.
      • + *
    + */ +public class UsageExemptBilling { + protected Logger logger = LogManager.getLogger(getClass()); + + /** What was reported for one tap last time, and the counts behind it. */ + static class NicState { + int ifindex; + long libvirtSent; + long libvirtReceived; + long exemptSent; + long exemptReceived; + long exemptSentBeforeReset; + long exemptReceivedBeforeReset; + long reportedSent; + long reportedReceived; + + NicState() { + } + + NicState(int ifindex) { + this.ifindex = ifindex; + } + } + + /** Exempt bytes counted on one tap. Bytes sent include the Ethernet header, the same as libvirt counts them. */ + public static class ExemptCount { + long sent; + long received; + + ExemptCount() { + } + + ExemptCount(long sent, long received) { + this.sent = sent; + this.received = received; + } + } + + private final Path stateFile; + private final Gson gson = new Gson(); + private Map nicStates; + + UsageExemptBilling(Path stateFile) { + this.stateFile = stateFile; + } + + /** Whether there's a record from earlier in this boot. */ + boolean hasSavedState() { + return Files.exists(stateFile); + } + + /** + * Takes exempt bytes off one tap's libvirt totals and returns the bytes to report as sent and received. + * Pass null for exempt when the counters couldn't be read. The tap then keeps its exempt bytes from last time. + */ + synchronized long[] subtractExempt(String device, int ifindex, long sent, long received, ExemptCount exempt) { + loadState(); + NicState nic = nicStates.get(device); + if (nic == null || nic.ifindex != ifindex || sent < nic.libvirtSent || received < nic.libvirtReceived) { + nic = new NicState(ifindex); + nicStates.put(device, nic); + } + if (exempt == null) { + exempt = new ExemptCount(nic.exemptSent, nic.exemptReceived); + } + if (exempt.sent < nic.exemptSent) { + nic.exemptSentBeforeReset += nic.exemptSent; + } + if (exempt.received < nic.exemptReceived) { + nic.exemptReceivedBeforeReset += nic.exemptReceived; + } + nic.exemptSent = exempt.sent; + nic.exemptReceived = exempt.received; + nic.libvirtSent = sent; + nic.libvirtReceived = received; + nic.reportedSent = billable(sent, exempt.sent + nic.exemptSentBeforeReset, nic.reportedSent); + nic.reportedReceived = billable(received, exempt.received + nic.exemptReceivedBeforeReset, nic.reportedReceived); + return new long[] {nic.reportedSent, nic.reportedReceived}; + } + + /** The total less exempt bytes, held between what was reported last time and the total. */ + static long billable(long total, long exempt, long reported) { + return Math.max(reported, Math.min(total, Math.max(0, total - exempt))); + } + + /** + * Called when a tap is getting a new counter for what it sends, receives or both, because the old one is gone. Keeps + * what the old counter had counted. By the next stats collection the new counter may have counted past it, and + * {@link #subtractExempt} would no longer see the drop. Returns true if anything was kept. + */ + synchronized boolean keepCountsFromLostChains(int ifindex, boolean sent, boolean received) { + if (!sent && !received) { + return false; + } + loadState(); + boolean keptCounts = false; + for (NicState nic : nicStates.values()) { + if (nic.ifindex != ifindex) { + continue; + } + if (sent && nic.exemptSent > 0) { + nic.exemptSentBeforeReset += nic.exemptSent; + nic.exemptSent = 0; + keptCounts = true; + } + if (received && nic.exemptReceived > 0) { + nic.exemptReceivedBeforeReset += nic.exemptReceived; + nic.exemptReceived = 0; + keptCounts = true; + } + } + return keptCounts; + } + + /** Writes the records to /run, dropping taps that no longer exist. tapExists gets each tap's name and ifindex. */ + synchronized void saveState(BiPredicate tapExists) { + if (nicStates == null) { + return; + } + nicStates.entrySet().removeIf(nic -> !tapExists.test(nic.getKey(), nic.getValue().ifindex)); + try { + Files.createDirectories(stateFile.getParent()); + final Path temp = stateFile.resolveSibling(stateFile.getFileName() + ".tmp"); + Files.write(temp, gson.toJson(nicStates).getBytes(StandardCharsets.UTF_8)); + Files.move(temp, stateFile, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE); + } catch (IOException e) { + logger.warn("Failed to save {}", stateFile, e); + } + } + + private void loadState() { + if (nicStates != null) { + return; + } + nicStates = new HashMap<>(); + if (!Files.exists(stateFile)) { + return; + } + try { + final Type type = new TypeToken>() { }.getType(); + final Map saved = gson.fromJson(new String(Files.readAllBytes(stateFile), StandardCharsets.UTF_8), type); + if (saved != null) { + nicStates.putAll(saved); + } + } catch (IOException | RuntimeException e) { + logger.warn("Failed to read {}. Starting afresh.", stateFile, e); + } + } +} diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/UsageExemptCounters.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/UsageExemptCounters.java new file mode 100644 index 000000000000..36be6a8cdc95 --- /dev/null +++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/UsageExemptCounters.java @@ -0,0 +1,478 @@ +// 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; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Stream; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; + +import com.cloud.hypervisor.kvm.resource.UsageExemptBilling.ExemptCount; +import com.cloud.utils.script.OutputInterpreter; +import com.cloud.utils.script.Script; + +/** + * Keeps traffic between VMs and the addresses in network.usage.exempt.cidrs out of the network usage this host reports. + *

    + * The usage we report comes from libvirt, which reads the byte counters on each VM NIC's tap device. Those are plain + * totals, so we put a second counter on the same tap that only counts exempt traffic, and report the total minus that. + * The counters are nftables rules in a netdev table. A tap's ingress hook sees what the VM sends and its egress hook sees + * what the VM receives. Hooking the tap works whichever bridge or switch it's plugged into. + *

    + * This class looks after the nftables side and reads the counts. {@link UsageExemptBilling} does the subtraction and keeps + * the reported numbers safe to bill. + *

      + *
    • The ingress hook runs after the kernel has read past the 14 byte Ethernet header, so it doesn't count it. The tap + * counters do, so we add it back for each packet the VM sends. The egress hook sees the whole frame already.
      • + *
    • Chains are named after the tap's ifindex, which the kernel doesn't reuse until a reboot. A new tap that gets an old + * name starts from zero, the same as libvirt's counters do.
      • + *
    • Only taps that belong to user VMs get counters. Usage from the host is only billed for VMs on shared networks, + * so system VMs such as virtual routers don't need them. User VMs on isolated and VPC networks still get counters, even + * though their usage comes from the virtual router. The host can't tell which kind of network a NIC is on. Telling it + * would mean sending extra NIC details from the management server when a VM starts, gets a new NIC or migrates.
      • + *
    + * The egress hook needs Linux 5.16 or later, or a kernel with it backported, such as EL9. On older kernels we log one + * warning and report the libvirt totals unchanged. + */ +public class UsageExemptCounters { + protected Logger logger = LogManager.getLogger(getClass()); + + static final String TABLE = "cloudstack_usage_exempt"; + static final int ETHERNET_HEADER_BYTES = 14; + static final long SYNC_INTERVAL_SECONDS = 10; + private static final String SYS_CLASS_NET = "/sys/class/net"; + private static final String STATE_FILE = "/run/cloudstack-agent/usage-exempt.json"; + private static final long NFT_TIMEOUT_MS = 30000; + private static final Pattern CHAIN_NAME = Pattern.compile("(in|out)_(\\d+)"); + private static final Pattern DEVICE_NAME = Pattern.compile("[A-Za-z0-9_.-]{1,15}"); + + private final Path netDir; + private final UsageExemptBilling billing; + + private List cidrs; + private List appliedCidrs; + private Map exemptCounts; + private ScheduledExecutorService periodicSync; + private Boolean egressSupported; + private boolean active; + private Supplier> tapOwners; + private final Map isUserVmTap = new HashMap<>(); + + /** Marks a stats collection where we couldn't read the counters. NICs keep the exempt counts from last time. */ + private static final Map COUNTS_UNREADABLE = new HashMap<>(); + + public UsageExemptCounters() { + this(Paths.get(SYS_CLASS_NET), Paths.get(STATE_FILE)); + } + + UsageExemptCounters(Path netDir, Path stateFile) { + this.netDir = netDir; + this.billing = new UsageExemptBilling(stateFile); + } + + /** + * Sets how we find which VM owns each tap: a map from NIC device name to VM name, or null if it can't be read. + */ + public synchronized void setTapOwners(Supplier> tapOwners) { + this.tapOwners = tapOwners; + } + + /** + * Takes a new exempt list from the management server and applies it straight away. The server sends it on every + * connect, empty when the setting isn't used. A host with an empty list and no table from before does nothing. + * Otherwise we also start a sync every {@link #SYNC_INTERVAL_SECONDS} seconds, which puts counters on new taps, such + * as a VM that just started or migrated here, and repairs anything that went missing. + */ + public synchronized void setCidrs(List cidrs) { + this.cidrs = new ArrayList<>(cidrs); + // An empty list only matters if this host has been counting since it booted. + active = !cidrs.isEmpty() || (billing.hasSavedState() && tableExists()); + if (!active) { + return; + } + try { + sync(); + } catch (RuntimeException e) { + logger.warn("Failed to set up the usage exemption counters. The next sync will try again.", e); + } + if (Boolean.FALSE.equals(egressSupported)) { + active = false; + return; + } + if (periodicSync == null && !cidrs.isEmpty()) { + periodicSync = startPeriodicSync(); + } + } + + protected ScheduledExecutorService startPeriodicSync() { + final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(r -> { + Thread thread = new Thread(r, "UsageExemptCounters"); + thread.setDaemon(true); + return thread; + }); + executor.scheduleWithFixedDelay(this::syncQuietly, SYNC_INTERVAL_SECONDS, SYNC_INTERVAL_SECONDS, TimeUnit.SECONDS); + return executor; + } + + private void syncQuietly() { + try { + synchronized (this) { + if (active && cidrs != null && !cidrs.isEmpty()) { + sync(); + } + } + } catch (RuntimeException e) { + logger.warn("Failed to sync the usage exemption counters", e); + } + } + + /** + * Reads our counters ahead of a stats collection, after making sure every tap has them. Call {@link #subtractExempt} for + * each NIC afterwards, then {@link #saveState}, all while holding this object's lock. + */ + public synchronized void readExemptCounts() { + if (!active && cidrs == null && billing.hasSavedState() && tableExists()) { + // The agent restarted and the server hasn't sent the list yet. The table kept counting, so keep using it. + // Reporting libvirt's totals now would jump above what we reported before, then drop once the list arrives. + active = true; + } + exemptCounts = active ? sync() : null; + } + + /** + * Brings the nftables table in line with the exempt list and the taps on this host, and returns the counters as they + * were before any change. Returns null when there's no table and we can't make one, and {@link #COUNTS_UNREADABLE} when the + * table can't be read. Until the server sends a list, it only removes stale chains. + */ + private Map sync() { + final String listing = listTable(); + if (listing == null && (cidrs == null || !isEgressSupported())) { + return null; + } + final Map exempt = new HashMap<>(); + final Map chains = new HashMap<>(); + if (listing != null) { + try { + readTable(listing, chains, exempt); + } catch (RuntimeException e) { + logger.warn("Couldn't read the {} nftables table. Keeping the exempt counts from last time.", TABLE, e); + return COUNTS_UNREADABLE; + } + } + final Map taps = keepUserVmTaps(listTaps()); + + // Sets first, in their own transaction, so a tap that vanishes mid-sync can't hold up a change to the list. + final StringBuilder sets = new StringBuilder(); + if (listing == null) { + sets.append("add table netdev ").append(TABLE).append('\n'); + sets.append("add set netdev ").append(TABLE).append(" v4 { type ipv4_addr; flags interval; auto-merge; }\n"); + sets.append("add set netdev ").append(TABLE).append(" v6 { type ipv6_addr; flags interval; auto-merge; }\n"); + appliedCidrs = null; + } + if (cidrs != null && !cidrs.equals(appliedCidrs)) { + appendSet(sets, "v4", false); + appendSet(sets, "v6", true); + } + if (sets.length() > 0) { + if (!applyRules(sets.toString())) { + return exempt; + } + appliedCidrs = cidrs; + } + + // Chains whose tap has gone. The kernel usually removes them with the tap, so this one can fail harmlessly. + final StringBuilder stale = new StringBuilder(); + for (Map.Entry chain : chains.entrySet()) { + final Matcher name = CHAIN_NAME.matcher(chain.getKey()); + final int ifindex = name.matches() ? Integer.parseInt(name.group(2)) : -1; + if (!chain.getValue().equals(taps.get(ifindex))) { + stale.append("delete chain netdev ").append(TABLE).append(' ').append(chain.getKey()).append('\n'); + exempt.remove(ifindex); + } + } + if (stale.length() > 0) { + applyRules(stale.toString()); + } + + if (cidrs != null && !cidrs.isEmpty()) { + final StringBuilder added = new StringBuilder(); + boolean keptCounts = false; + for (Map.Entry tap : taps.entrySet()) { + final boolean in = appendChain(added, chains, "in", "ingress", "daddr", tap.getKey(), tap.getValue()); + final boolean out = appendChain(added, chains, "out", "egress", "saddr", tap.getKey(), tap.getValue()); + keptCounts |= billing.keepCountsFromLostChains(tap.getKey(), in, out); + } + if (keptCounts) { + saveState(); + } + if (added.length() > 0 && !applyRules(added.toString())) { + logger.warn("Failed to add {} nftables chains for new NICs. Their exempt traffic is billed until the next sync works.", TABLE); + } + } + return exempt; + } + + /** + * Keeps the taps that belong to user VMs. We ask libvirt once per new tap and remember the answer. A tap libvirt + * doesn't know yet is asked about again on the next sync. + */ + private Map keepUserVmTaps(Map taps) { + if (tapOwners == null) { + return taps; + } + isUserVmTap.keySet().retainAll(taps.keySet()); + if (!isUserVmTap.keySet().containsAll(taps.keySet())) { + final Map owners = tapOwners.get(); + if (owners != null) { + for (Map.Entry tap : taps.entrySet()) { + final String owner = owners.get(tap.getValue()); + if (owner != null) { + isUserVmTap.putIfAbsent(tap.getKey(), owner.startsWith("i-")); + } + } + } + } + final Map userTaps = new HashMap<>(); + for (Map.Entry tap : taps.entrySet()) { + if (Boolean.TRUE.equals(isUserVmTap.get(tap.getKey()))) { + userTaps.put(tap.getKey(), tap.getValue()); + } + } + return userTaps; + } + + /** + * Checks once whether the kernel supports the netdev egress hook, with a dry run that changes nothing. + */ + private boolean isEgressSupported() { + if (egressSupported == null && !isNftInstalled()) { + logger.warn("The nft command isn't installed (package nftables). " + + "network.usage.exempt.cidrs won't apply to VMs on this host, and their usage stays as libvirt reports it."); + egressSupported = false; + } + if (egressSupported == null) { + egressSupported = checkRules("table netdev " + TABLE + "_probe {\n" + + " chain probe { type filter hook egress device \"lo\" priority 0; }\n}\n"); + if (!egressSupported) { + logger.warn("This kernel doesn't support the nftables netdev egress hook (Linux 5.16 or later, or EL9). " + + "network.usage.exempt.cidrs won't apply to VMs on this host, and their usage stays as libvirt reports it."); + } + } + return egressSupported; + } + + private void appendSet(StringBuilder script, String set, boolean ipv6) { + final List family = new ArrayList<>(); + for (String cidr : cidrs) { + if (cidr.contains(":") == ipv6) { + family.add(cidr); + } + } + script.append("flush set netdev ").append(TABLE).append(' ').append(set).append('\n'); + if (!family.isEmpty()) { + script.append("add element netdev ").append(TABLE).append(' ').append(set) + .append(" { ").append(String.join(", ", family)).append(" }\n"); + } + } + + /** Adds a chain for one tap and direction unless it's there already. Returns true if it adds one. */ + private boolean appendChain(StringBuilder script, Map chains, String prefix, String hook, String match, + int ifindex, String device) { + final String chain = prefix + "_" + ifindex; + if (chains.containsKey(chain)) { + return false; + } + final String rule = "add rule netdev " + TABLE + " " + chain + " "; + script.append("add chain netdev ").append(TABLE).append(' ').append(chain) + .append(" { type filter hook ").append(hook).append(" device \"").append(device).append("\" priority 0; }\n"); + script.append(rule).append("ip ").append(match).append(" @v4 counter\n"); + script.append(rule).append("ip6 ").append(match).append(" @v6 counter\n"); + return true; + } + + /** Fills in our chains (name to device) and the exempt counters per ifindex from an nft JSON listing. */ + private void readTable(String listing, Map chains, Map exempt) { + final JsonArray items = JsonParser.parseString(listing).getAsJsonObject().getAsJsonArray("nftables"); + for (JsonElement item : items) { + final JsonObject object = item.getAsJsonObject(); + if (object.has("chain")) { + final JsonObject chain = object.getAsJsonObject("chain"); + chains.put(chain.get("name").getAsString(), chain.has("dev") ? chain.get("dev").getAsString() : ""); + } else if (object.has("rule")) { + final JsonObject rule = object.getAsJsonObject("rule"); + final Matcher name = CHAIN_NAME.matcher(rule.get("chain").getAsString()); + if (!name.matches()) { + continue; + } + final ExemptCount counted = exempt.computeIfAbsent(Integer.parseInt(name.group(2)), k -> new ExemptCount()); + for (JsonElement expr : rule.getAsJsonArray("expr")) { + final JsonObject counter = expr.getAsJsonObject().getAsJsonObject("counter"); + if (counter == null) { + continue; + } + final long bytes = counter.get("bytes").getAsLong(); + final long packets = counter.get("packets").getAsLong(); + if ("in".equals(name.group(1))) { + counted.sent += bytes + packets * ETHERNET_HEADER_BYTES; + } else { + counted.received += bytes; + } + } + } + } + } + + /** The tap devices on this host, by ifindex. */ + private Map listTaps() { + final Map taps = new HashMap<>(); + try (Stream devices = Files.list(netDir)) { + devices.forEach(device -> { + final String name = device.getFileName().toString(); + if (DEVICE_NAME.matcher(name).matches() && Files.exists(device.resolve("tun_flags"))) { + final Integer ifindex = readIfindex(name); + if (ifindex != null) { + taps.put(ifindex, name); + } + } + }); + } catch (IOException e) { + logger.warn("Failed to list the network devices in {}", netDir, e); + } + return taps; + } + + private Integer readIfindex(String device) { + if (device == null || !DEVICE_NAME.matcher(device).matches()) { + return null; + } + try { + return Integer.valueOf(new String(Files.readAllBytes(netDir.resolve(device).resolve("ifindex")), StandardCharsets.UTF_8).trim()); + } catch (IOException | NumberFormatException e) { + return null; + } + } + + /** + * Takes the exempt traffic off one NIC's libvirt totals. Returns the bytes to report as sent and received. + * NICs without a kernel network device, such as vhost-user ports, are returned unchanged. + */ + public synchronized long[] subtractExempt(String device, long sent, long received) { + final Integer ifindex = readIfindex(device); + if (exemptCounts == null || ifindex == null) { + return new long[] {sent, received}; + } + final ExemptCount exempt = exemptCounts == COUNTS_UNREADABLE ? null : exemptCounts.getOrDefault(ifindex, new ExemptCount()); + return billing.subtractExempt(device, ifindex, sent, received, exempt); + } + + /** Writes the billing record of each tap to /run, dropping taps that no longer exist. */ + public synchronized void saveState() { + billing.saveState((device, ifindex) -> ifindex.equals(readIfindex(device))); + } + + protected boolean isNftInstalled() { + final Script nft = new Script("nft", NFT_TIMEOUT_MS, logger); + nft.add("--version"); + return nft.execute() == null; + } + + /** + * Checks for our table with a listing that succeeds either way. Script logs a warning for any failed command, + * so we don't ask nft for a table that isn't there. + */ + protected boolean tableExists() { + final Script nft = new Script("nft", NFT_TIMEOUT_MS, logger); + nft.add("list", "tables", "netdev"); + final OutputInterpreter.AllLinesParser output = new OutputInterpreter.AllLinesParser(); + return nft.execute(output) == null && output.getLines() != null + && Pattern.compile("^table netdev " + TABLE + "$", Pattern.MULTILINE).matcher(output.getLines()).find(); + } + + /** Lists our table as JSON, or returns null if it doesn't exist. */ + protected String listTable() { + if (!tableExists()) { + return null; + } + final Script nft = new Script("nft", NFT_TIMEOUT_MS, logger); + nft.add("-j", "list", "table", "netdev", TABLE); + final OutputInterpreter.AllLinesParser output = new OutputInterpreter.AllLinesParser(); + return nft.execute(output) == null ? output.getLines() : null; + } + + /** Applies nft rules in one transaction. */ + protected boolean applyRules(String rules) { + return runRulesFile(rules, false); + } + + /** Dry-runs nft rules. The kernel validates them and nothing changes. */ + protected boolean checkRules(String rules) { + return runRulesFile(rules, true); + } + + /** + * Script closes a command's input as soon as it starts it, so the rules can't be piped into nft. + * They're written to a temp file, and nft reads them from there. + */ + private boolean runRulesFile(String rules, boolean checkOnly) { + Path file = null; + try { + file = Files.createTempFile("cloudstack-usage-exempt", ".nft"); + Files.write(file, rules.getBytes(StandardCharsets.UTF_8)); + final Script nft = new Script("nft", NFT_TIMEOUT_MS, logger); + if (checkOnly) { + nft.add("-c"); + } + nft.add("-f", file.toString()); + final String error = nft.execute(); + if (error != null && !checkOnly) { + logger.warn("Failed to apply the {} nftables rules: {}", TABLE, error); + } + return error == null; + } catch (IOException e) { + logger.warn("Failed to write the {} nftables rules to a temp file", TABLE, e); + return false; + } finally { + if (file != null) { + try { + Files.deleteIfExists(file); + } catch (IOException e) { + logger.debug("Failed to delete {}", file, e); + } + } + } + } +} diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtGetVmNetworkStatsCommandWrapper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtGetVmNetworkStatsCommandWrapper.java index 03321122c4ae..b32e9f06a461 100644 --- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtGetVmNetworkStatsCommandWrapper.java +++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtGetVmNetworkStatsCommandWrapper.java @@ -30,6 +30,7 @@ import com.cloud.agent.api.GetVmNetworkStatsCommand; import com.cloud.agent.api.VmNetworkStatsEntry; import com.cloud.hypervisor.kvm.resource.LibvirtComputingResource; +import com.cloud.hypervisor.kvm.resource.UsageExemptCounters; import com.cloud.resource.CommandWrapper; import com.cloud.resource.ResourceWrapper; @@ -45,17 +46,22 @@ public Answer execute(final GetVmNetworkStatsCommand command, final LibvirtCompu try { final HashMap> vmNetworkStatsNameMap = new HashMap>(); final Connect conn = libvirtUtilitiesHelper.getConnection(); - for (final String vmName : vmNames) { - try { - final List statEntry = libvirtComputingResource.getVmNetworkStat(conn, vmName); - if (statEntry == null) { - continue; - } + final UsageExemptCounters usageExemptCounters = libvirtComputingResource.getUsageExemptCounters(); + synchronized (usageExemptCounters) { + usageExemptCounters.readExemptCounts(); + for (final String vmName : vmNames) { + try { + final List statEntry = libvirtComputingResource.getVmNetworkStat(conn, vmName); + if (statEntry == null) { + continue; + } - vmNetworkStatsNameMap.put(vmName, statEntry); - } catch (LibvirtException e) { - logger.warn("Can't get vm network stats: " + e.toString() + ", continue"); + vmNetworkStatsNameMap.put(vmName, statEntry); + } catch (LibvirtException e) { + logger.warn("Can't get vm network stats: " + e.toString() + ", continue"); + } } + usageExemptCounters.saveState(); } return new GetVmNetworkStatsAnswer(command, "", command.getHostName(), vmNetworkStatsNameMap); } catch (final LibvirtException e) { diff --git a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/UsageExemptBillingTest.java b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/UsageExemptBillingTest.java new file mode 100644 index 000000000000..9b71585f74d7 --- /dev/null +++ b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/UsageExemptBillingTest.java @@ -0,0 +1,131 @@ +// 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; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Collections; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.cloud.hypervisor.kvm.resource.UsageExemptBilling.ExemptCount; + +public class UsageExemptBillingTest { + + private Path dir; + private Path stateFile; + private UsageExemptBilling billing; + + @Before + public void setUp() throws IOException { + dir = Files.createTempDirectory("usage-exempt-billing"); + stateFile = dir.resolve("run/state.json"); + billing = new UsageExemptBilling(stateFile); + } + + @After + public void tearDown() throws IOException { + try (java.util.stream.Stream paths = Files.walk(dir)) { + paths.sorted(Collections.reverseOrder()).forEach(path -> path.toFile().delete()); + } + } + + private long[] report(long sent, long received, long exemptSent, long exemptReceived) { + return billing.subtractExempt("vnet1", 51, sent, received, new ExemptCount(exemptSent, exemptReceived)); + } + + @Test + public void subtractsExemptBytesFromBothDirections() { + Assert.assertArrayEquals(new long[] {4000, 7300}, report(5000, 8000, 1000, 700)); + } + + @Test + public void neverReportsLessThanLastTime() { + Assert.assertEquals(4000, report(5000, 0, 1000, 0)[0]); + // exempt traffic counted by nft but not yet by libvirt would push the figure down + Assert.assertEquals(4000, report(5500, 0, 2000, 0)[0]); + Assert.assertEquals(4500, report(6500, 0, 2000, 0)[0]); + } + + @Test + public void keepsExemptBytesWhenTheCountGoesDown() { + Assert.assertEquals(7000, report(0, 10000, 0, 3000)[1]); + // the counter was wiped and started again + Assert.assertEquals(12000 - 3000 - 500, report(0, 12000, 0, 500)[1]); + } + + @Test + public void keepCountsFromLostChainsCatchesAResetTheNextReportWouldMiss() { + Assert.assertEquals(4000, report(5000, 0, 1000, 0)[0]); + Assert.assertTrue(billing.keepCountsFromLostChains(51, true, false)); + // the new counter has already counted past the old one + Assert.assertEquals(7000 - 1000 - 1500, report(7000, 0, 1500, 0)[0]); + } + + @Test + public void keepCountsFromLostChainsIgnoresOtherTaps() { + report(5000, 0, 1000, 0); + Assert.assertFalse(billing.keepCountsFromLostChains(52, true, true)); + } + + @Test + public void newIfindexStartsAfresh() { + Assert.assertEquals(9000, report(10000, 0, 1000, 0)[0]); + Assert.assertEquals(300, billing.subtractExempt("vnet1", 77, 300, 0, new ExemptCount())[0]); + } + + @Test + public void libvirtTotalsGoingDownStartsAfresh() { + Assert.assertEquals(9000, report(10000, 0, 1000, 0)[0]); + Assert.assertEquals(300, report(300, 0, 0, 0)[0]); + } + + @Test + public void unreadableCountsKeepTheExemptBytesFromLastTime() { + Assert.assertEquals(4000, report(5000, 0, 1000, 0)[0]); + Assert.assertEquals(5000, billing.subtractExempt("vnet1", 51, 6000, 0, null)[0]); + } + + @Test + public void recordSurvivesAnAgentRestart() { + Assert.assertEquals(4000, report(5000, 0, 1000, 0)[0]); + billing.saveState((device, ifindex) -> true); + Assert.assertTrue(new UsageExemptBilling(stateFile).hasSavedState()); + + UsageExemptBilling restarted = new UsageExemptBilling(stateFile); + Assert.assertEquals(4000, restarted.subtractExempt("vnet1", 51, 5500, 0, new ExemptCount(2000, 0))[0]); + } + + @Test + public void saveStateDropsTapsThatAreGone() { + report(5000, 0, 1000, 0); + billing.saveState((device, ifindex) -> false); + UsageExemptBilling restarted = new UsageExemptBilling(stateFile); + Assert.assertEquals(5500 - 2000, restarted.subtractExempt("vnet1", 51, 5500, 0, new ExemptCount(2000, 0))[0]); + } + + @Test + public void billableStaysBetweenLastReportAndTotal() { + Assert.assertEquals(700, UsageExemptBilling.billable(1000, 300, 0)); + Assert.assertEquals(800, UsageExemptBilling.billable(1000, 300, 800)); + Assert.assertEquals(0, UsageExemptBilling.billable(1000, 5000, 0)); + } +} diff --git a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/UsageExemptCountersTest.java b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/UsageExemptCountersTest.java new file mode 100644 index 000000000000..9af8b285b73f --- /dev/null +++ b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/UsageExemptCountersTest.java @@ -0,0 +1,370 @@ +// 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; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class UsageExemptCountersTest { + + /** Stands in for nft: keeps the chains and counters we set, and records each script it's given. */ + static class FakeNftCounters extends UsageExemptCounters { + boolean tableExists; + boolean failApply; + String failIfContains; + String listingOverride; + boolean egressSupported = true; + boolean nftInstalled = true; + int checks; + final Map chains = new LinkedHashMap<>(); + final Map counters = new LinkedHashMap<>(); + final List scripts = new ArrayList<>(); + + FakeNftCounters(Path netDir, Path stateFile) { + super(netDir, stateFile); + } + + @Override + protected ScheduledExecutorService startPeriodicSync() { + return null; + } + + void count(String chain, long packets, long bytes) { + counters.put(chain, new long[] {packets, bytes}); + } + + @Override + protected boolean tableExists() { + return tableExists; + } + + @Override + protected String listTable() { + if (!tableExists) { + return null; + } + if (listingOverride != null) { + return listingOverride; + } + StringBuilder json = new StringBuilder("{\"nftables\": [{\"metainfo\": {}}, {\"table\": {\"family\": \"netdev\", \"name\": \"t\"}}"); + for (Map.Entry chain : chains.entrySet()) { + json.append(", {\"chain\": {\"name\": \"").append(chain.getKey()).append("\", \"dev\": \"").append(chain.getValue()).append("\"}}"); + long[] counter = counters.getOrDefault(chain.getKey(), new long[] {0, 0}); + json.append(", {\"rule\": {\"chain\": \"").append(chain.getKey()).append("\", \"expr\": [{\"match\": {}}, {\"counter\": {\"packets\": ") + .append(counter[0]).append(", \"bytes\": ").append(counter[1]).append("}}]}}"); + json.append(", {\"rule\": {\"chain\": \"").append(chain.getKey()).append("\", \"expr\": [{\"match\": {}}, {\"counter\": {\"packets\": 0, \"bytes\": 0}}]}}"); + } + return json.append("]}").toString(); + } + + @Override + protected boolean applyRules(String script) { + scripts.add(script); + if (failApply || (failIfContains != null && script.contains(failIfContains))) { + return false; + } + tableExists = true; + for (String line : script.split("\n")) { + String[] words = line.split(" "); + if (line.startsWith("add chain")) { + chains.put(words[4], line.replaceAll(".*device \"([^\"]+)\".*", "$1")); + } else if (line.startsWith("delete chain")) { + chains.remove(words[4]); + counters.remove(words[4]); + } + } + return true; + } + + @Override + protected boolean isNftInstalled() { + return nftInstalled; + } + + @Override + protected boolean checkRules(String rules) { + checks++; + return egressSupported; + } + + String lastScript() { + return scripts.get(scripts.size() - 1); + } + + long[] collect(String device, long sent, long received) { + readExemptCounts(); + long[] usage = subtractExempt(device, sent, received); + saveState(); + return usage; + } + } + + private Path dir; + private Path netDir; + private FakeNftCounters counters; + + @Before + public void setUp() throws IOException { + dir = Files.createTempDirectory("usage-exempt"); + netDir = dir.resolve("net"); + counters = new FakeNftCounters(netDir, dir.resolve("run/state.json")); + addTap("vnet1", 51); + } + + @After + public void tearDown() throws IOException { + try (java.util.stream.Stream paths = Files.walk(dir)) { + paths.sorted(Collections.reverseOrder()).forEach(path -> path.toFile().delete()); + } + } + + private void addTap(String name, int ifindex) throws IOException { + Path device = netDir.resolve(name); + Files.createDirectories(device); + Files.write(device.resolve("ifindex"), (ifindex + "\n").getBytes(StandardCharsets.UTF_8)); + Files.write(device.resolve("tun_flags"), "0x1002\n".getBytes(StandardCharsets.UTF_8)); + } + + private void removeTap(String name) throws IOException { + Path device = netDir.resolve(name); + Files.delete(device.resolve("ifindex")); + Files.delete(device.resolve("tun_flags")); + Files.delete(device); + } + + @Test + public void reportsLibvirtTotalsWhenNeverConfigured() { + Assert.assertArrayEquals(new long[] {1000, 2000}, counters.collect("vnet1", 1000, 2000)); + Assert.assertTrue(counters.scripts.isEmpty()); + } + + @Test + public void setCidrsCreatesTheTableSetsAndChains() { + counters.setCidrs(Arrays.asList("192.0.2.5/32", "2001:db8::/64", "198.51.100.0/24")); + String script = String.join("", counters.scripts); + Assert.assertTrue(script.contains("add table netdev cloudstack_usage_exempt\n")); + Assert.assertTrue(script.contains("add element netdev cloudstack_usage_exempt v4 { 192.0.2.5/32, 198.51.100.0/24 }\n")); + Assert.assertTrue(script.contains("add element netdev cloudstack_usage_exempt v6 { 2001:db8::/64 }\n")); + Assert.assertTrue(script.contains("add chain netdev cloudstack_usage_exempt in_51 { type filter hook ingress device \"vnet1\" priority 0; }\n")); + Assert.assertTrue(script.contains("add rule netdev cloudstack_usage_exempt in_51 ip daddr @v4 counter\n")); + Assert.assertTrue(script.contains("add rule netdev cloudstack_usage_exempt out_51 ip6 saddr @v6 counter\n")); + } + + @Test + public void unchangedListIsNotReapplied() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + int applied = counters.scripts.size(); + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + Assert.assertEquals(applied, counters.scripts.size()); + counters.setCidrs(Collections.singletonList("192.0.2.6/32")); + Assert.assertEquals(applied + 1, counters.scripts.size()); + Assert.assertTrue(counters.lastScript().contains("add element")); + Assert.assertFalse(counters.lastScript().contains("add chain")); + } + + @Test + public void subtractsExemptTrafficAndAddsTheEthernetHeaderToWhatTheVmSent() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.count("in_51", 10, 1000); + counters.count("out_51", 5, 700); + Assert.assertArrayEquals(new long[] {5000 - 1000 - 10 * 14, 8000 - 700}, counters.collect("vnet1", 5000, 8000)); + } + + @Test + public void carriesExemptBytesWhenTheChainsAreRecreated() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.count("out_51", 0, 3000); + Assert.assertEquals(7000, counters.collect("vnet1", 0, 10000)[1]); + // someone flushes nftables and the sync puts the table back with fresh counters + counters.tableExists = false; + counters.chains.clear(); + counters.counters.clear(); + counters.count("out_51", 0, 500); + counters.tableExists = true; + counters.chains.put("in_51", "vnet1"); + counters.chains.put("out_51", "vnet1"); + Assert.assertEquals(12000 - 3000 - 500, counters.collect("vnet1", 0, 12000)[1]); + } + + @Test + public void newTapWithAnOldNameStartsAfresh() throws IOException { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.count("in_51", 0, 1000); + Assert.assertEquals(9000, counters.collect("vnet1", 10000, 0)[0]); + removeTap("vnet1"); + addTap("vnet1", 77); + counters.readExemptCounts(); + Assert.assertFalse(counters.chains.containsKey("in_51")); + Assert.assertEquals("vnet1", counters.chains.get("in_77")); + Assert.assertEquals(300, counters.collect("vnet1", 300, 0)[0]); + } + + @Test + public void stateSurvivesAnAgentRestart() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.count("in_51", 0, 1000); + Assert.assertEquals(4000, counters.collect("vnet1", 5000, 0)[0]); + + FakeNftCounters restarted = new FakeNftCounters(netDir, dir.resolve("run/state.json")); + restarted.tableExists = true; + restarted.chains.putAll(counters.chains); + restarted.count("in_51", 0, 2000); + // the agent reconnects and gets the same list; the table kept counting while it was down + restarted.setCidrs(Collections.singletonList("192.0.2.5/32")); + Assert.assertFalse(restarted.lastScript().contains("add chain")); + Assert.assertEquals(4000, restarted.collect("vnet1", 5500, 0)[0]); + } + + @Test + public void clearingTheListKeepsWhatWasAlreadyExempt() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.count("in_51", 0, 1000); + Assert.assertEquals(4000, counters.collect("vnet1", 5000, 0)[0]); + counters.setCidrs(Collections.emptyList()); + Assert.assertTrue(counters.lastScript().contains("flush set netdev cloudstack_usage_exempt v4\n")); + Assert.assertFalse(counters.lastScript().contains("add element")); + Assert.assertEquals(5000, counters.collect("vnet1", 6000, 0)[0]); + } + + @Test + public void missingNftReportsLibvirtTotals() { + counters.nftInstalled = false; + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + Assert.assertTrue(counters.scripts.isEmpty()); + Assert.assertEquals("skips the kernel check", 0, counters.checks); + Assert.assertArrayEquals(new long[] {1000, 2000}, counters.collect("vnet1", 1000, 2000)); + } + + @Test + public void emptyListWithNoTableDoesNothing() { + counters.setCidrs(Collections.emptyList()); + Assert.assertArrayEquals(new long[] {1000, 2000}, counters.collect("vnet1", 1000, 2000)); + Assert.assertTrue(counters.scripts.isEmpty()); + Assert.assertEquals(0, counters.checks); + } + + @Test + public void nicWithoutAKernelDeviceIsReportedUnchanged() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + Assert.assertArrayEquals(new long[] {1000, 2000}, counters.collect("vhost-user-port", 1000, 2000)); + } + + @Test + public void failedApplyIsRetriedOnTheNextSync() { + counters.failApply = true; + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.failApply = false; + int failed = counters.scripts.size(); + counters.readExemptCounts(); + Assert.assertEquals("vnet1", counters.chains.get("in_51")); + Assert.assertTrue(String.join("", counters.scripts.subList(failed, counters.scripts.size())).contains("add element")); + } + + @Test + public void kernelWithoutTheEgressHookReportsLibvirtTotals() { + counters.egressSupported = false; + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + Assert.assertTrue(counters.scripts.isEmpty()); + Assert.assertArrayEquals(new long[] {1000, 2000}, counters.collect("vnet1", 1000, 2000)); + Assert.assertEquals("checks the kernel once", 1, counters.checks); + } + + @Test + public void restartBeforeTheListArrivesKeepsUsingTheTable() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.count("in_51", 0, 1000); + Assert.assertEquals(4000, counters.collect("vnet1", 5000, 0)[0]); + + FakeNftCounters restarted = new FakeNftCounters(netDir, dir.resolve("run/state.json")); + restarted.tableExists = true; + restarted.chains.putAll(counters.chains); + restarted.count("in_51", 0, 2000); + // a stats request arrives before the server has sent the list + Assert.assertEquals(4000, restarted.collect("vnet1", 5500, 0)[0]); + Assert.assertEquals(4500, restarted.collect("vnet1", 6500, 0)[0]); + Assert.assertTrue(restarted.scripts.isEmpty()); + } + + @Test + public void flushRepairedBetweenStatsCollectionsKeepsTheCarry() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.count("in_51", 0, 1000); + Assert.assertEquals(4000, counters.collect("vnet1", 5000, 0)[0]); + // someone flushes nftables, and the background sync puts the table back + counters.tableExists = false; + counters.chains.clear(); + counters.counters.clear(); + counters.readExemptCounts(); + Assert.assertEquals("vnet1", counters.chains.get("in_51")); + // by the next stats collection the new chain has counted past the old one + counters.count("in_51", 0, 1500); + Assert.assertEquals(7000 - 1000 - 1500, counters.collect("vnet1", 7000, 0)[0]); + } + + @Test + public void unreadableTableKeepsTheExemptCountsFromLastTime() { + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + counters.count("in_51", 0, 1000); + Assert.assertEquals(4000, counters.collect("vnet1", 5000, 0)[0]); + counters.listingOverride = "{\"nftables\": [{\"rule\": {\"chain\": \"in_51\", \"expr\": \"surprise\"}}]}"; + Assert.assertEquals(5000, counters.collect("vnet1", 6000, 0)[0]); + } + + @Test + public void systemVmTapsGetNoCounters() throws IOException { + addTap("vnet2", 52); + int[] lookups = {0}; + Map owners = new LinkedHashMap<>(); + owners.put("vnet1", "i-2-10-VM"); + owners.put("vnet2", "r-11-VM"); + counters.setTapOwners(() -> { + lookups[0]++; + return owners; + }); + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + Assert.assertEquals("vnet1", counters.chains.get("in_51")); + Assert.assertFalse(counters.chains.containsKey("in_52")); + counters.readExemptCounts(); + Assert.assertEquals("asks libvirt once per new tap", 1, lookups[0]); + } + + @Test + public void listChangesEvenWhenATapVanishesMidSync() { + counters.failIfContains = "add chain"; + counters.setCidrs(Collections.singletonList("192.0.2.5/32")); + Assert.assertTrue(counters.scripts.get(0).contains("add element netdev cloudstack_usage_exempt v4 { 192.0.2.5/32 }")); + Assert.assertTrue(counters.chains.isEmpty()); + counters.failIfContains = null; + counters.readExemptCounts(); + Assert.assertEquals("vnet1", counters.chains.get("in_51")); + Assert.assertFalse("the list isn't applied again", counters.lastScript().contains("add element")); + } +} diff --git a/server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java b/server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java index 6b2af820fc2e..867a7a8ba958 100644 --- a/server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java +++ b/server/src/main/java/com/cloud/configuration/ConfigurationManagerImpl.java @@ -692,6 +692,8 @@ public void onPublishMessage(String senderAddress, String subject, Object args) params.put(Config.RouterAggregationCommandEachTimeout.toString(), _configDao.getValue(Config.RouterAggregationCommandEachTimeout.toString())); params.put(Config.MigrateWait.toString(), _configDao.getValue(Config.MigrateWait.toString())); _agentManager.propagateChangeToAgents(params); + } else if (settingNameUpdated.equals(NetworkOrchestrationService.NetworkUsageExemptCidrs.key())) { + _agentManager.propagateUsageExemptCidrsToAgents(settingUpdated.second() == ConfigKey.Scope.Zone ? settingUpdated.third() : null); } else if (settingNameUpdated.equals(IndirectAgentLBServiceImpl.IndirectAgentLBCheckInterval.key())) { ConfigKey.Scope scope = settingUpdated.second(); if (scope == ConfigKey.Scope.Global) { @@ -1348,6 +1350,11 @@ public Pair resetConfiguration(final ResetCfgCmd cmd) thr _configDepot.invalidateConfigCache(name, scope, id); + // Resetting doesn't publish the change event that updateConfiguration does, so tell the hosts here. + if (NetworkOrchestrationService.NetworkUsageExemptCidrs.key().equals(name)) { + _agentManager.propagateUsageExemptCidrsToAgents(scope == ConfigKey.Scope.Zone ? id : null); + } + CallContext.current().setEventDetails(" Name: " + name + " New Value: " + (name.toLowerCase().contains("password") ? "*****" : defaultValue == null ? "" : defaultValue)); return new Pair<>(_configDao.findByName(name), newValue); }