Linstor: add HA support and small cleanups (#8407)

* linstor: Outline get storagepools from resourcegroup into function

* linstor: move getHostname() to kvm/Pool and reimplement

* linstor: implement CloudStack HA support

plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/storage/LinstorStorageAdaptor.java

@@ -16,18 +16,16 @@
 // under the License.
 package com.cloud.hypervisor.kvm.storage;
 
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.StringJoiner;
 
 import javax.annotation.Nonnull;
 
+import com.cloud.storage.Storage;
+import com.cloud.utils.exception.CloudRuntimeException;
 import org.apache.cloudstack.storage.datastore.util.LinstorUtil;
 import org.apache.cloudstack.utils.qemu.QemuImg;
 import org.apache.cloudstack.utils.qemu.QemuImgException;
@@ -35,8 +33,6 @@ import org.apache.cloudstack.utils.qemu.QemuImgFile;
 import org.apache.log4j.Logger;
 import org.libvirt.LibvirtException;
 
-import com.cloud.storage.Storage;
-import com.cloud.utils.exception.CloudRuntimeException;
 import com.linbit.linstor.api.ApiClient;
 import com.linbit.linstor.api.ApiException;
 import com.linbit.linstor.api.Configuration;
@@ -47,7 +43,6 @@ import com.linbit.linstor.api.model.Properties;
 import com.linbit.linstor.api.model.ProviderKind;
 import com.linbit.linstor.api.model.ResourceDefinition;
 import com.linbit.linstor.api.model.ResourceDefinitionModify;
-import com.linbit.linstor.api.model.ResourceGroup;
 import com.linbit.linstor.api.model.ResourceGroupSpawn;
 import com.linbit.linstor.api.model.ResourceMakeAvailable;
 import com.linbit.linstor.api.model.ResourceWithVolumes;
@@ -70,28 +65,6 @@ public class LinstorStorageAdaptor implements StorageAdaptor {
         return LinstorUtil.RSC_PREFIX + name;
     }
 
-    private String getHostname() {
-        // either there is already some function for that in the agent or a better way.
-        ProcessBuilder pb = new ProcessBuilder("hostname");
-        try
-        {
-            String result;
-            Process p = pb.start();
-            final BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
-
-            StringJoiner sj = new StringJoiner(System.getProperty("line.separator"));
-            reader.lines().iterator().forEachRemaining(sj::add);
-            result = sj.toString();
-
-            p.waitFor();
-            p.destroy();
-            return result.trim();
-        } catch (IOException | InterruptedException exc) {
-            Thread.currentThread().interrupt();
-            throw new CloudRuntimeException("Unable to run 'hostname' command.");
-        }
-    }
-
     private void logLinstorAnswer(@Nonnull ApiCallRc answer) {
         if (answer.isError()) {
             s_logger.error(answer.getMessage());
@@ -122,7 +95,7 @@ public class LinstorStorageAdaptor implements StorageAdaptor {
     }
 
     public LinstorStorageAdaptor() {
-        localNodeName = getHostname();
+        localNodeName = LinstorStoragePool.getHostname();
     }
 
     @Override
@@ -511,25 +484,7 @@ public class LinstorStorageAdaptor implements StorageAdaptor {
         DevelopersApi linstorApi = getLinstorAPI(pool);
         final String rscGroupName = pool.getResourceGroup();
         try {
-            List<ResourceGroup> rscGrps = linstorApi.resourceGroupList(
-                Collections.singletonList(rscGroupName),
-                null,
-                null,
-                null);
-
-            if (rscGrps.isEmpty()) {
-                final String errMsg = String.format("Linstor: Resource group '%s' not found", rscGroupName);
-                s_logger.error(errMsg);
-                throw new CloudRuntimeException(errMsg);
-            }
-
-            List<StoragePool> storagePools = linstorApi.viewStoragePools(
-                Collections.emptyList(),
-                rscGrps.get(0).getSelectFilter().getStoragePoolList(),
-                null,
-                null,
-                null
-            );
+            List<StoragePool> storagePools = LinstorUtil.getRscGroupStoragePools(linstorApi, rscGroupName);
 
             final long free = storagePools.stream()
                 .filter(sp -> sp.getProviderKind() != ProviderKind.DISKLESS)
@@ -547,25 +502,7 @@ public class LinstorStorageAdaptor implements StorageAdaptor {
         DevelopersApi linstorApi = getLinstorAPI(pool);
         final String rscGroupName = pool.getResourceGroup();
         try {
-            List<ResourceGroup> rscGrps = linstorApi.resourceGroupList(
-                Collections.singletonList(rscGroupName),
-                null,
-                null,
-                null);
-
-            if (rscGrps.isEmpty()) {
-                final String errMsg = String.format("Linstor: Resource group '%s' not found", rscGroupName);
-                s_logger.error(errMsg);
-                throw new CloudRuntimeException(errMsg);
-            }
-
-            List<StoragePool> storagePools = linstorApi.viewStoragePools(
-                Collections.emptyList(),
-                rscGrps.get(0).getSelectFilter().getStoragePoolList(),
-                null,
-                null,
-                null
-            );
+            List<StoragePool> storagePools = LinstorUtil.getRscGroupStoragePools(linstorApi, rscGroupName);
 
             final long used = storagePools.stream()
                 .filter(sp -> sp.getProviderKind() != ProviderKind.DISKLESS)

plugins/storage/volume/linstor/src/main/java/com/cloud/hypervisor/kvm/storage/LinstorStoragePool.java

@@ -19,20 +19,33 @@ package com.cloud.hypervisor.kvm.storage;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.cloudstack.utils.qemu.QemuImg;
-import org.joda.time.Duration;
-
 import com.cloud.agent.api.to.HostTO;
+import com.cloud.agent.properties.AgentProperties;
+import com.cloud.agent.properties.AgentPropertiesFileHandler;
 import com.cloud.hypervisor.kvm.resource.KVMHABase.HAStoragePool;
 import com.cloud.storage.Storage;
+import com.cloud.utils.exception.CloudRuntimeException;
+import com.cloud.utils.script.OutputInterpreter;
+import com.cloud.utils.script.Script;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonIOException;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import com.google.gson.JsonSyntaxException;
+import org.apache.cloudstack.utils.qemu.QemuImg;
+import org.apache.log4j.Logger;
+import org.joda.time.Duration;
 
 public class LinstorStoragePool implements KVMStoragePool {
+    private static final Logger s_logger = Logger.getLogger(LinstorStoragePool.class);
     private final String _uuid;
     private final String _sourceHost;
     private final int _sourcePort;
     private final Storage.StoragePoolType _storagePoolType;
     private final StorageAdaptor _storageAdaptor;
     private final String _resourceGroup;
+    private final String localNodeName;
 
     public LinstorStoragePool(String uuid, String host, int port, String resourceGroup,
                               Storage.StoragePoolType storagePoolType, StorageAdaptor storageAdaptor) {
@@ -42,6 +55,7 @@ public class LinstorStoragePool implements KVMStoragePool {
         _storagePoolType = storagePoolType;
         _storageAdaptor = storageAdaptor;
         _resourceGroup = resourceGroup;
+        localNodeName = getHostname();
     }
 
     @Override
@@ -200,32 +214,132 @@ public class LinstorStoragePool implements KVMStoragePool {
 
     @Override
     public boolean isPoolSupportHA() {
-        return false;
+        return true;
     }
 
     @Override
     public String getHearthBeatPath() {
-        return null;
+        String kvmScriptsDir = AgentPropertiesFileHandler.getPropertyValue(AgentProperties.KVM_SCRIPTS_DIR);
+        return Script.findScript(kvmScriptsDir, "kvmspheartbeat.sh");
     }
 
     @Override
-    public String createHeartBeatCommand(HAStoragePool primaryStoragePool, String hostPrivateIp,
+    public String createHeartBeatCommand(HAStoragePool pool, String hostPrivateIp,
             boolean hostValidation) {
-        return null;
+        s_logger.trace(String.format("Linstor.createHeartBeatCommand: %s, %s, %b", pool.getPoolIp(), hostPrivateIp, hostValidation));
+        boolean isStorageNodeUp = checkingHeartBeat(pool, null);
+        if (!isStorageNodeUp && !hostValidation) {
+            //restart the host
+            s_logger.debug(String.format("The host [%s] will be restarted because the health check failed for the storage pool [%s]", hostPrivateIp, pool.getPool().getType()));
+            Script cmd = new Script(pool.getPool().getHearthBeatPath(), Duration.millis(HeartBeatUpdateTimeout), s_logger);
+            cmd.add("-c");
+            cmd.execute();
+            return "Down";
+        }
+        return isStorageNodeUp ? null : "Down";
     }
 
     @Override
     public String getStorageNodeId() {
+        // only called by storpool
         return null;
     }
 
+    static String getHostname() {
+        OutputInterpreter.AllLinesParser parser = new OutputInterpreter.AllLinesParser();
+        Script sc = new Script("hostname", Duration.millis(10000L), s_logger);
+        String res = sc.execute(parser);
+        if (res != null) {
+            throw new CloudRuntimeException(String.format("Unable to run 'hostname' command: %s", res));
+        }
+        String response = parser.getLines();
+        return response.trim();
+    }
+
     @Override
     public Boolean checkingHeartBeat(HAStoragePool pool, HostTO host) {
-        return null;
+        String hostName;
+        if (host == null) {
+            hostName = localNodeName;
+        } else {
+            hostName = host.getParent();
+            if (hostName == null) {
+                s_logger.error("No hostname set in host.getParent()");
+                return false;
+            }
+        }
+
+        return checkHostUpToDateAndConnected(hostName);
+    }
+
+    private String executeDrbdSetupStatus(OutputInterpreter.AllLinesParser parser) {
+        Script sc = new Script("drbdsetup", Duration.millis(HeartBeatUpdateTimeout), s_logger);
+        sc.add("status");
+        sc.add("--json");
+        return sc.execute(parser);
+    }
+
+    private boolean checkDrbdSetupStatusOutput(String output, String otherNodeName) {
+        JsonParser jsonParser = new JsonParser();
+        JsonArray jResources = (JsonArray) jsonParser.parse(output);
+        for (JsonElement jElem : jResources) {
+            JsonObject jRes = (JsonObject) jElem;
+            JsonArray jConnections = jRes.getAsJsonArray("connections");
+            for (JsonElement jConElem : jConnections) {
+                JsonObject jConn = (JsonObject) jConElem;
+                if (jConn.getAsJsonPrimitive("name").getAsString().equals(otherNodeName)
+                        && jConn.getAsJsonPrimitive("connection-state").getAsString().equalsIgnoreCase("Connected")) {
+                    return true;
+                }
+            }
+        }
+        s_logger.warn(String.format("checkDrbdSetupStatusOutput: no resource connected to %s.", otherNodeName));
+        return false;
+    }
+
+    private String executeDrbdEventsNow(OutputInterpreter.AllLinesParser parser) {
+        Script sc = new Script("drbdsetup", Duration.millis(HeartBeatUpdateTimeout), s_logger);
+        sc.add("events2");
+        sc.add("--now");
+        return sc.execute(parser);
+    }
+
+    private boolean checkDrbdEventsNowOutput(String output) {
+        boolean healthy = output.lines().noneMatch(line -> line.matches(".*role:Primary .* promotion_score:0.*"));
+        if (!healthy) {
+            s_logger.warn("checkDrbdEventsNowOutput: primary resource with promotion score==0; HA false");
+        }
+        return healthy;
+    }
+
+    private boolean checkHostUpToDateAndConnected(String hostName) {
+        s_logger.trace(String.format("checkHostUpToDateAndConnected: %s/%s", localNodeName, hostName));
+        OutputInterpreter.AllLinesParser parser = new OutputInterpreter.AllLinesParser();
+
+        if (localNodeName.equalsIgnoreCase(hostName)) {
+            String res = executeDrbdEventsNow(parser);
+            if (res != null) {
+                return false;
+            }
+            return checkDrbdEventsNowOutput(parser.getLines());
+        } else {
+            // check drbd connections
+            String res = executeDrbdSetupStatus(parser);
+            if (res != null) {
+                return false;
+            }
+            try {
+                return checkDrbdSetupStatusOutput(parser.getLines(), hostName);
+            } catch (JsonIOException | JsonSyntaxException e) {
+                s_logger.error("Error parsing drbdsetup status --json", e);
+            }
+        }
+        return false;
     }
 
     @Override
     public Boolean vmActivityCheck(HAStoragePool pool, HostTO host, Duration activityScriptTimeout, String volumeUUIDListString, String vmActivityCheckPath, long duration) {
-        return null;
+        s_logger.trace(String.format("Linstor.vmActivityCheck: %s, %s", pool.getPoolIp(), host.getPrivateNetwork().getIp()));
+        return checkingHeartBeat(pool, host);
     }
 }

plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/driver/LinstorPrimaryDataStoreDriverImpl.java

@@ -1241,7 +1241,7 @@ public class LinstorPrimaryDataStoreDriverImpl implements PrimaryDataStoreDriver
 
     @Override
     public boolean isStorageSupportHA(StoragePoolType type) {
-        return false;
+        return true;
     }
 
     @Override

plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/provider/LinstorHostListener.java

@@ -0,0 +1,32 @@
+// 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.storage.datastore.provider;
+
+import com.cloud.exception.StorageConflictException;
+import com.cloud.host.HostVO;
+
+public class LinstorHostListener extends DefaultHostListener {
+    @Override
+    public boolean hostConnect(long hostId, long poolId) throws StorageConflictException {
+        HostVO host = hostDao.findById(hostId);
+        if (host.getParent() == null) {
+            host.setParent(host.getName());
+            hostDao.update(host.getId(), host);
+        }
+        return super.hostConnect(hostId, poolId);
+    }
+}

plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/provider/LinstorPrimaryDatastoreProviderImpl.java

@@ -48,7 +48,7 @@ public class LinstorPrimaryDatastoreProviderImpl implements PrimaryDataStoreProv
     public boolean configure(Map<String, Object> params) {
         lifecycle = ComponentContext.inject(LinstorPrimaryDataStoreLifeCycleImpl.class);
         driver = ComponentContext.inject(LinstorPrimaryDataStoreDriverImpl.class);
-        listener = ComponentContext.inject(DefaultHostListener.class);
+        listener = ComponentContext.inject(LinstorHostListener.class);
         return true;
     }
 

plugins/storage/volume/linstor/src/main/java/org/apache/cloudstack/storage/datastore/util/LinstorUtil.java

@@ -136,28 +136,33 @@ public class LinstorUtil {
         return path;
     }
 
-    public static long getCapacityBytes(String linstorUrl, String rscGroupName) {
-        DevelopersApi linstorApi = getLinstorAPI(linstorUrl);
-        try {
-            List<ResourceGroup> rscGrps = linstorApi.resourceGroupList(
+    public static List<StoragePool> getRscGroupStoragePools(DevelopersApi api, String rscGroupName)
+            throws ApiException {
+        List<ResourceGroup> rscGrps = api.resourceGroupList(
                 Collections.singletonList(rscGroupName),
                 null,
                 null,
                 null);
 
-            if (rscGrps.isEmpty()) {
-                final String errMsg = String.format("Linstor: Resource group '%s' not found", rscGroupName);
-                s_logger.error(errMsg);
-                throw new CloudRuntimeException(errMsg);
-            }
+        if (rscGrps.isEmpty()) {
+            final String errMsg = String.format("Linstor: Resource group '%s' not found", rscGroupName);
+            s_logger.error(errMsg);
+            throw new CloudRuntimeException(errMsg);
+        }
 
-            List<StoragePool> storagePools = linstorApi.viewStoragePools(
+        return api.viewStoragePools(
                 Collections.emptyList(),
                 rscGrps.get(0).getSelectFilter().getStoragePoolList(),
                 null,
                 null,
                 null
-            );
+        );
+    }
+
+    public static long getCapacityBytes(String linstorUrl, String rscGroupName) {
+        DevelopersApi linstorApi = getLinstorAPI(linstorUrl);
+        try {
+            List<StoragePool> storagePools = getRscGroupStoragePools(linstorApi, rscGroupName);
 
             return storagePools.stream()
                 .filter(sp -> sp.getProviderKind() != ProviderKind.DISKLESS)