vr: serve UDP load balancer rules with IPVS

PendingReleaseNotes

@@ -39,3 +39,13 @@ example.ver.1 > example.ver.2:
     which can now be attached to Instances. This is to prevent the Secondary
     Storage to grow to enormous sizes as Linux Distributions keep growing in
     size while a stripped down Linux should fit on a 2.88MB floppy.
+
+
+4.22.0.0 > 4.23.0.0:
+  * KVM/Ceph: RBD volumes can now be encrypted at rest using native librbd
+    LUKS2 (<encryption format='luks2' engine='librbd'>), for both data disks
+    and root disks. Encryption is transparent to the guest and reuses the
+    existing CloudStack volume-encryption passphrase handling, so no
+    additional key store is required. Note: attaching an encrypted RBD volume
+    to a running Instance requires libvirt >= 10.1.0; booting an Instance from
+    an encrypted RBD root disk works on older libvirt.

agent/src/main/java/com/cloud/agent/Agent.java

@@ -969,6 +969,7 @@ private Answer migrateAgentToOtherMS(final MigrateAgentConnectionCommand cmd) {
             }
             ScheduledExecutorService migrateAgentConnectionService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("MigrateAgentConnection-Job"));
             migrateAgentConnectionService.schedule(() -> {
+                waitForCommandsToFinish(MIGRATE_AGENT_CONNECTION_MAX_WAIT_SECS);
                 migrateAgentConnection(cmd.getAvoidMsList());
             }, 3, TimeUnit.SECONDS);
             migrateAgentConnectionService.shutdown();
@@ -980,6 +981,37 @@ private Answer migrateAgentToOtherMS(final MigrateAgentConnectionCommand cmd) {
         return new MigrateAgentConnectionAnswer(true);
     }
 
+    private static final int MIGRATE_AGENT_CONNECTION_MAX_WAIT_SECS = 300;
+
+    /**
+     * Answers travel back over the current link, so a command still running or queued when the agent reconnects to
+     * another management server would lose its answer and fail there. Wait until nothing has run or been queued for two
+     * consecutive checks; the management server holds new commands for this host while it is Rebalancing.
+     */
+    private void waitForCommandsToFinish(final int maxWaitSecs) {
+        final long deadline = System.currentTimeMillis() + maxWaitSecs * 1000L;
+        int idleChecks = 0;
+        while (idleChecks < 2) {
+            final int queued = requestHandler instanceof ThreadPoolExecutor ? ((ThreadPoolExecutor)requestHandler).getQueue().size() : 0;
+            idleChecks = (commandsInProgress.get() == 0 && queued == 0) ? idleChecks + 1 : 0;
+            if (idleChecks >= 2) {
+                break;
+            }
+            if (System.currentTimeMillis() >= deadline) {
+                logger.warn("Moving the agent connection with {} commands still in progress and {} queued after {} s",
+                        commandsInProgress.get(), queued, maxWaitSecs);
+                return;
+            }
+            try {
+                Thread.sleep(1000);
+            } catch (final InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return;
+            }
+        }
+        logger.info("No commands in progress, moving the agent connection");
+    }
+
     private void migrateAgentConnection(List<String> avoidMsList) {
         final String[] msHosts = shell.getHosts();
         if (msHosts == null || msHosts.length < 1) {

api/src/main/java/com/cloud/host/Host.java

@@ -56,6 +56,7 @@ public static String[] toStrings(Host.Type... types) {
 
     String HOST_UEFI_ENABLE = "host.uefi.enable";
     String HOST_VOLUME_ENCRYPTION = "host.volume.encryption";
+    String HOST_RBD_VOLUME_ENCRYPTION = "host.volume.encryption.rbd";
     String HOST_INSTANCE_CONVERSION = "host.instance.conversion";
     String HOST_VDDK_SUPPORT = "host.vddk.support";
     String HOST_VDDK_LIB_DIR = "vddk.lib.dir";

api/src/main/java/com/cloud/storage/Storage.java

@@ -172,7 +172,7 @@ public static enum StoragePoolType {
         LVM(false, false, EncryptionSupport.Unsupported), // XenServer local LVM SR
         CLVM(true, false, EncryptionSupport.Unsupported),
         CLVM_NG(true, false, EncryptionSupport.Hypervisor),
-        RBD(true, true, EncryptionSupport.Unsupported), // http://libvirt.org/storage.html#StorageBackendRBD
+        RBD(true, true, EncryptionSupport.Hypervisor), // http://libvirt.org/storage.html#StorageBackendRBD ; encrypted natively by librbd (LUKS2, engine='librbd')
         SharedMountPoint(true, true, EncryptionSupport.Hypervisor),
         VMFS(true, true, EncryptionSupport.Unsupported), // VMware VMFS storage
         PreSetup(true, true, EncryptionSupport.Unsupported), // for XenServer, Storage Pool is set up by customers.

api/src/main/java/org/apache/cloudstack/api/command/user/loadbalancer/CreateLoadBalancerRuleCmd.java

@@ -327,6 +327,10 @@ public Integer getSourcePortEnd() {
     }
 
     public String getProtocol() {
+        // a UDP load balancer needs a UDP firewall rule and must not clash with UDP port forwarding on the same port
+        if (NetUtils.UDP_PROTO.equals(getLbProtocol())) {
+            return NetUtils.UDP_PROTO;
+        }
         return NetUtils.TCP_PROTO;
     }
 

core/src/main/java/com/cloud/agent/resource/virtualnetwork/facade/LoadBalancerConfigItem.java

@@ -58,6 +58,7 @@ public List<ConfigItem> generateConfig(final NetworkElementCommand cmd) {
         final LoadBalancerRule loadBalancerRule = new LoadBalancerRule(configuration, tmpCfgFilePath, tmpCfgFileName, addRules, removeRules, statRules, routerIp);
         final LoadBalancerRule.SslCertEntry[] sslCerts = cfgtr.generateSslCertEntries(command);
         loadBalancerRule.setSslCerts(sslCerts);
+        loadBalancerRule.setUdpRules(cfgtr.generateUdpEntries(command));
 
         final List<LoadBalancerRule> rules = new LinkedList<LoadBalancerRule>();
         rules.add(loadBalancerRule);

core/src/main/java/com/cloud/agent/resource/virtualnetwork/model/LoadBalancerRule.java

@@ -26,6 +26,7 @@ public class LoadBalancerRule {
     private String tmpCfgFilePath;
     private String tmpCfgFileName;
     private SslCertEntry[] sslCerts;
+    private UdpLbEntry[] udpRules;
 
     private String[] addRules;
     private String[] removeRules;
@@ -80,6 +81,50 @@ public String getPassword() {
         }
     }
 
+    /**
+     * A UDP load balancer rule. The router serves it with IPVS (NAT mode), not haproxy.
+     */
+    public static class UdpLbEntry {
+        private String sourceIp;
+        private int sourcePort;
+        // IPVS scheduler: rr, lc or sh
+        private String scheduler;
+        // IPVS persistence timeout in seconds, 0 = none
+        private int persistence;
+        // "ip:port" of each backend
+        private String[] destinations;
+        // allowed source CIDRs, empty = everyone
+        private String[] cidrs;
+
+        public UdpLbEntry(String sourceIp, int sourcePort, String scheduler, int persistence, String[] destinations, String[] cidrs) {
+            this.sourceIp = sourceIp;
+            this.sourcePort = sourcePort;
+            this.scheduler = scheduler;
+            this.persistence = persistence;
+            this.destinations = destinations;
+            this.cidrs = cidrs;
+        }
+
+        public String getSourceIp() {
+            return sourceIp;
+        }
+        public int getSourcePort() {
+            return sourcePort;
+        }
+        public String getScheduler() {
+            return scheduler;
+        }
+        public int getPersistence() {
+            return persistence;
+        }
+        public String[] getDestinations() {
+            return destinations;
+        }
+        public String[] getCidrs() {
+            return cidrs;
+        }
+    }
+
     public LoadBalancerRule() {
         // Empty constructor for (de)serialization
     }
@@ -157,4 +202,12 @@ public SslCertEntry[] getSslCerts() {
     public void setSslCerts(final SslCertEntry[] sslCerts) {
         this.sslCerts = sslCerts;
     }
+
+    public UdpLbEntry[] getUdpRules() {
+        return udpRules;
+    }
+
+    public void setUdpRules(final UdpLbEntry[] udpRules) {
+        this.udpRules = udpRules;
+    }
 }

core/src/main/java/com/cloud/network/HAProxyConfigurator.java

@@ -37,6 +37,7 @@
 import com.cloud.agent.api.to.LoadBalancerTO.StickinessPolicyTO;
 import com.cloud.agent.api.to.PortForwardingRuleTO;
 import com.cloud.agent.resource.virtualnetwork.model.LoadBalancerRule.SslCertEntry;
+import com.cloud.agent.resource.virtualnetwork.model.LoadBalancerRule.UdpLbEntry;
 import com.cloud.network.lb.LoadBalancingRule.LbSslCert;
 import com.cloud.network.rules.LbStickinessMethod.StickinessMethodType;
 import com.cloud.utils.Pair;
@@ -683,7 +684,7 @@ public String[] generateConfiguration(final LoadBalancerConfigCommand lbCmd) {
         result.add(blankLine);
         boolean has_listener = false;
         for (final LoadBalancerTO lbTO : lbCmd.getLoadBalancers()) {
-            if (lbTO.isRevoked()) {
+            if (lbTO.isRevoked() || isUdp(lbTO)) {
                 continue;
             }
             final List<String> poolRules = getRulesForPool(lbTO, lbCmd);
@@ -708,6 +709,9 @@ public String[][] generateFwRules(final LoadBalancerConfigCommand lbCmd) {
         final Set<String> toStats = new HashSet<String>();
 
         for (final LoadBalancerTO lbTO : lbCmd.getLoadBalancers()) {
+            if (isUdp(lbTO)) {
+                continue;
+            }
 
             final StringBuilder sb = new StringBuilder();
             sb.append(lbTO.getSrcIp()).append(":");
@@ -764,4 +768,73 @@ private void addSslCertEntry(Set<SslCertEntry> sslCertEntries, LoadBalancerTO lb
         final SslCertEntry sslCertEntry = new SslCertEntry(name, cert.getCert(), cert.getKey(), cert.getChain(), cert.getPassword());
         sslCertEntries.add(sslCertEntry);
     }
+
+    private static boolean isUdp(final LoadBalancerTO lbTO) {
+        return NetUtils.UDP_PROTO.equalsIgnoreCase(lbTO.getLbProtocol());
+    }
+
+    @Override
+    public UdpLbEntry[] generateUdpEntries(final LoadBalancerConfigCommand lbCmd) {
+        final List<UdpLbEntry> result = new ArrayList<>();
+        for (final LoadBalancerTO lbTO : lbCmd.getLoadBalancers()) {
+            if (lbTO.isRevoked() || !isUdp(lbTO)) {
+                continue;
+            }
+            final List<String> destinations = new ArrayList<>();
+            for (final DestinationTO dest : lbTO.getDestinations()) {
+                if (!dest.isRevoked()) {
+                    destinations.add(dest.getDestIp() + ":" + dest.getDestPort());
+                }
+            }
+            final String[] cidrs = StringUtils.isBlank(lbTO.getCidrList()) ? new String[0] : lbTO.getCidrList().trim().split("\\s+");
+            result.add(new UdpLbEntry(lbTO.getSrcIp(), lbTO.getSrcPort(), getIpvsScheduler(lbTO.getAlgorithm()), getIpvsPersistence(lbTO),
+                    destinations.toArray(new String[0]), cidrs));
+        }
+        return result.toArray(new UdpLbEntry[0]);
+    }
+
+    protected static String getIpvsScheduler(final String algorithm) {
+        if ("leastconn".equals(algorithm)) {
+            return "lc";
+        }
+        if ("source".equals(algorithm)) {
+            return "sh";
+        }
+        return "rr";
+    }
+
+    /**
+     * Source based stickiness becomes IPVS persistence; its "expire" value (default 3h, as for haproxy) is the timeout.
+     */
+    protected static int getIpvsPersistence(final LoadBalancerTO lbTO) {
+        if (lbTO.getStickinessPolicies() == null) {
+            return 0;
+        }
+        for (final StickinessPolicyTO policy : lbTO.getStickinessPolicies()) {
+            if (policy == null || !StickinessMethodType.SourceBased.getName().equalsIgnoreCase(policy.getMethodName())) {
+                continue;
+            }
+            String expire = "3h";
+            for (final Pair<String, String> param : policy.getParams()) {
+                if ("expire".equalsIgnoreCase(param.first()) && StringUtils.isNotBlank(param.second())) {
+                    expire = param.second().trim();
+                }
+            }
+            return durationToSeconds(expire);
+        }
+        return 0;
+    }
+
+    private static int durationToSeconds(final String duration) {
+        final char unit = duration.charAt(duration.length() - 1);
+        final int multiplier;
+        switch (unit) {
+            case 's': multiplier = 1; break;
+            case 'm': multiplier = 60; break;
+            case 'h': multiplier = 3600; break;
+            case 'd': multiplier = 86400; break;
+            default: return Integer.parseInt(duration);
+        }
+        return Integer.parseInt(duration.substring(0, duration.length() - 1)) * multiplier;
+    }
 }

core/src/main/java/com/cloud/network/LoadBalancerConfigurator.java

@@ -24,6 +24,7 @@
 import com.cloud.agent.api.routing.LoadBalancerConfigCommand;
 import com.cloud.agent.api.to.PortForwardingRuleTO;
 import com.cloud.agent.resource.virtualnetwork.model.LoadBalancerRule.SslCertEntry;
+import com.cloud.agent.resource.virtualnetwork.model.LoadBalancerRule.UdpLbEntry;
 
 public interface LoadBalancerConfigurator {
     public final static int ADD = 0;
@@ -37,4 +38,6 @@ public interface LoadBalancerConfigurator {
     public String[][] generateFwRules(LoadBalancerConfigCommand lbCmd);
 
     public SslCertEntry[] generateSslCertEntries(LoadBalancerConfigCommand lbCmd);
+
+    public UdpLbEntry[] generateUdpEntries(LoadBalancerConfigCommand lbCmd);
 }

core/src/main/java/org/apache/cloudstack/backup/RestoreBackupCommand.java

@@ -24,7 +24,9 @@
 import com.cloud.vm.VirtualMachine;
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 
+import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 public class RestoreBackupCommand extends Command  {
     private String vmName;
@@ -40,11 +42,38 @@ public class RestoreBackupCommand extends Command  {
     private Boolean vmExists;
     private VirtualMachine.State vmState;
     private Integer mountTimeout;
+    /**
+     * Passphrase of every encrypted volume being restored, keyed by volume path. The backup itself is
+     * LUKS-encrypted (see TakeBackupCommand), so the agent needs these both to read the backup and to
+     * write the data back through the volume's own encryption.
+     */
+    @LogLevel(LogLevel.Log4jLevel.Off)
+    private Map<String, byte[]> volumePassphrases;
 
     protected RestoreBackupCommand() {
         super();
     }
 
+    public Map<String, byte[]> getVolumePassphrases() {
+        return volumePassphrases;
+    }
+
+    public void setVolumePassphrases(Map<String, byte[]> volumePassphrases) {
+        this.volumePassphrases = volumePassphrases;
+    }
+
+    /** Wipes the passphrases once the command has been handled. */
+    public void clearPassphrases() {
+        if (volumePassphrases == null) {
+            return;
+        }
+        for (byte[] passphrase : volumePassphrases.values()) {
+            if (passphrase != null) {
+                Arrays.fill(passphrase, (byte) 0);
+            }
+        }
+    }
+
     public String getVmName() {
         return vmName;
     }

core/src/main/java/org/apache/cloudstack/backup/TakeBackupCommand.java

@@ -23,7 +23,9 @@
 import com.cloud.agent.api.LogLevel;
 import org.apache.cloudstack.storage.to.PrimaryDataStoreTO;
 
+import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 public class TakeBackupCommand extends Command {
     private String vmName;
@@ -35,6 +37,15 @@ public class TakeBackupCommand extends Command {
     private Boolean quiesce;
     @LogLevel(LogLevel.Log4jLevel.Off)
     private String mountOptions;
+    /**
+     * Passphrase of every encrypted volume of the VM, keyed by volume path (the uuid the backup
+     * files are named after). Volumes that are not encrypted are absent. The agent uses these to
+     * make the backup itself LUKS-encrypted, so an encrypted volume is never written out in the
+     * clear. Keyed rather than positional because the running-VM path enumerates disks from
+     * libvirt instead of from {@link #volumePaths}.
+     */
+    @LogLevel(LogLevel.Log4jLevel.Off)
+    private Map<String, byte[]> volumePassphrases;
 
     // Incremental backup fields (NAS provider; null/empty for legacy full-only callers).
     private String mode;          // "full" or "incremental"; null => legacy behaviour (script default)
@@ -53,6 +64,26 @@ public TakeBackupCommand(String vmName, String backupPath) {
         this.backupPath = backupPath;
     }
 
+    public Map<String, byte[]> getVolumePassphrases() {
+        return volumePassphrases;
+    }
+
+    public void setVolumePassphrases(Map<String, byte[]> volumePassphrases) {
+        this.volumePassphrases = volumePassphrases;
+    }
+
+    /** Wipes the passphrases once the command has been handled. */
+    public void clearPassphrases() {
+        if (volumePassphrases == null) {
+            return;
+        }
+        for (byte[] passphrase : volumePassphrases.values()) {
+            if (passphrase != null) {
+                Arrays.fill(passphrase, (byte) 0);
+            }
+        }
+    }
+
     public String getVmName() {
         return vmName;
     }