implement generic retry function, use it to make tests more robust

Signed-off-by: Travis Glenn Hansen <travisghansen@yahoo.com>

ci/bin/run.ps1

@@ -58,7 +58,9 @@ while (!(Test-Path "${env:CSI_ENDPOINT}")) {
   $iter++
   Write-Output "Waiting for ${env:CSI_ENDPOINT} to appear"
   Start-Sleep 1
-  Get-Job | Receive-Job
+  try {
+    Get-Job | Receive-Job
+  } catch {}
   if ($iter -gt $max_iter) {
     Write-Output "${env:CSI_ENDPOINT} failed to appear"
     $started = 0
@@ -80,13 +82,18 @@ while ($csi_sanity_job -and ($csi_sanity_job.State -eq "Running" -or $csi_sanity
     if (($job -eq $csi_grpc_proxy_job) -and ($iter -gt 20)) {
       continue
     }
+    if (!$job.HasMoreData) {
+      continue
+    }
     try {
       $job | Receive-Job
     }
     catch {
       if ($job.State -ne "Failed") {
-        Write-Output "failure receiving job data: " + $_
+        Write-Output "failure receiving job data: ${_}"
-        $job | fl
+        # just swallow the errors as it seems there are various reasons errors
+        # may show up (perhaps no data currently, etc)
+        #$job | fl
         #throw $_
       }
     }

src/driver/controller-zfs-generic/index.js

@@ -1,9 +1,9 @@
 const _ = require("lodash");
 const { ControllerZfsBaseDriver } = require("../controller-zfs");
 const { GrpcError, grpc } = require("../../utils/grpc");
+const GeneralUtils = require("../../utils/general");
 const registry = require("../../utils/registry");
 const SshClient = require("../../utils/ssh").SshClient;
-const sleep = require("../../utils/general").sleep;
 const { Zetabyte, ZfsSshProcessManager } = require("../../utils/zfs");
 
 const Handlebars = require("handlebars");
@@ -231,8 +231,12 @@ class ControllerZfsGenericDriver extends ControllerZfsBaseDriver {
               }
             }
 
-            response = await this.targetCliCommand(
+            await GeneralUtils.retry(
-              `
+              3,
+              2000,
+              async () => {
+                await this.targetCliCommand(
+                  `
 # create target
 cd /iscsi
 create ${basename}:${iscsiName}
@@ -250,6 +254,16 @@ create ${iscsiName} /dev/${extentDiskName}
 cd /iscsi/${basename}:${iscsiName}/tpg1/luns
 create /backstores/block/${iscsiName}
 `
+                );
+              },
+              {
+                retryCondition: (err) => {
+                  if (err.stdout && err.stdout.includes("Ran out of input")) {
+                    return true;
+                  }
+                  return false;
+                },
+              }
             );
             break;
           default:
@@ -313,7 +327,7 @@ create /backstores/block/${iscsiName}
                 }
               }
             }
-            await sleep(2000); // let things settle
+            await GeneralUtils.sleep(2000); // let things settle
             break;
           default:
             throw new GrpcError(
@@ -343,7 +357,7 @@ create /backstores/block/${iscsiName}
                 }
               }
             }
-            await sleep(2000); // let things settle
+            await GeneralUtils.sleep(2000); // let things settle
             break;
           default:
             throw new GrpcError(
@@ -392,8 +406,12 @@ create /backstores/block/${iscsiName}
         switch (this.options.iscsi.shareStrategy) {
           case "targetCli":
             basename = this.options.iscsi.shareStrategyTargetCli.basename;
-            response = await this.targetCliCommand(
+            await GeneralUtils.retry(
-              `
+              3,
+              2000,
+              async () => {
+                await this.targetCliCommand(
+                  `
 # delete target
 cd /iscsi
 delete ${basename}:${iscsiName}
@@ -402,7 +420,18 @@ delete ${basename}:${iscsiName}
 cd /backstores/block
 delete ${iscsiName}
 `
+                );
+              },
+              {
+                retryCondition: (err) => {
+                  if (err.stdout && err.stdout.includes("Ran out of input")) {
+                    return true;
+                  }
+                  return false;
+                },
+              }
             );
+
             break;
           default:
             break;

src/driver/controller-zfs/index.js

@@ -1318,30 +1318,24 @@ class ControllerZfsBaseDriver extends CsiBaseDriver {
     // NOTE: -R will recursively delete items + dependent filesets
     // delete dataset
     try {
-      let max_tries = 12;
+      await GeneralUtils.retry(
-      let sleep_time = 5000;
+        12,
-      let current_try = 1;
+        5000,
-      let success = false;
+        async () => {
-      while (!success && current_try <= max_tries) {
-        try {
           await zb.zfs.destroy(datasetName, { recurse: true, force: true });
-          success = true;
+        },
-        } catch (err) {
+        {
-          if (
+          retryCondition: (err) => {
-            err.toString().includes("dataset is busy") ||
+            if (
-            err.toString().includes("target is busy")
+              err.toString().includes("dataset is busy") ||
-          ) {
+              err.toString().includes("target is busy")
-            current_try++;
+            ) {
-            if (current_try > max_tries) {
+              return true;
-              throw err;
-            } else {
-              await GeneralUtils.sleep(sleep_time);
             }
-          } else {
+            return false;
-            throw err;
+          },
-          }
         }
-      }
+      );
     } catch (err) {
       if (err.toString().includes("filesystem has dependent clones")) {
         throw new GrpcError(

src/driver/freenas/api.js

@@ -261,7 +261,27 @@ class FreeNASApiDriver extends CsiBaseDriver {
                   break;
               }
 
-              response = await httpClient.post("/sharing/nfs", share);
+              response = await GeneralUtils.retry(
+                3,
+                1000,
+                async () => {
+                  return await httpClient.post("/sharing/nfs", share);
+                },
+                {
+                  retryCondition: (err) => {
+                    if (err.code == "ECONNRESET") {
+                      return true;
+                    }
+                    if (err.code == "ECONNABORTED") {
+                      return true;
+                    }
+                    if (err.response && err.response.statusCode == 504) {
+                      return true;
+                    }
+                    return false;
+                  },
+                }
+              );
 
               /**
                * v1 = 201
@@ -482,7 +502,27 @@ class FreeNASApiDriver extends CsiBaseDriver {
                   break;
               }
 
-              response = await httpClient.post(endpoint, share);
+              response = await GeneralUtils.retry(
+                3,
+                1000,
+                async () => {
+                  return await httpClient.post(endpoint, share);
+                },
+                {
+                  retryCondition: (err) => {
+                    if (err.code == "ECONNRESET") {
+                      return true;
+                    }
+                    if (err.code == "ECONNABORTED") {
+                      return true;
+                    }
+                    if (err.response && err.response.statusCode == 504) {
+                      return true;
+                    }
+                    return false;
+                  },
+                }
+              );
 
               /**
                * v1 = 201

src/driver/freenas/ssh.js

@@ -6,7 +6,7 @@ const SshClient = require("../../utils/ssh").SshClient;
 const HttpClient = require("./http").Client;
 const TrueNASApiClient = require("./http/api").Api;
 const { Zetabyte, ZfsSshProcessManager } = require("../../utils/zfs");
-const { sleep, stringify } = require("../../utils/general");
+const GeneralUtils = require("../../utils/general");
 
 const Handlebars = require("handlebars");
 
@@ -308,7 +308,27 @@ class FreeNASSshDriver extends ControllerZfsBaseDriver {
                   break;
               }
 
-              response = await httpClient.post("/sharing/nfs", share);
+              response = await GeneralUtils.retry(
+                3,
+                1000,
+                async () => {
+                  return await httpClient.post("/sharing/nfs", share);
+                },
+                {
+                  retryCondition: (err) => {
+                    if (err.code == "ECONNRESET") {
+                      return true;
+                    }
+                    if (err.code == "ECONNABORTED") {
+                      return true;
+                    }
+                    if (err.response && err.response.statusCode == 504) {
+                      return true;
+                    }
+                    return false;
+                  },
+                }
+              );
 
               /**
                * v1 = 201
@@ -529,7 +549,27 @@ class FreeNASSshDriver extends ControllerZfsBaseDriver {
                   break;
               }
 
-              response = await httpClient.post(endpoint, share);
+              response = await GeneralUtils.retry(
+                3,
+                1000,
+                async () => {
+                  return await httpClient.post(endpoint, share);
+                },
+                {
+                  retryCondition: (err) => {
+                    if (err.code == "ECONNRESET") {
+                      return true;
+                    }
+                    if (err.code == "ECONNABORTED") {
+                      return true;
+                    }
+                    if (err.response && err.response.statusCode == 504) {
+                      return true;
+                    }
+                    return false;
+                  },
+                }
+              );
 
               /**
                * v1 = 201
@@ -1614,7 +1654,7 @@ class FreeNASSshDriver extends ControllerZfsBaseDriver {
                       targetId,
                       retries
                     );
-                    await sleep(retryWait);
+                    await GeneralUtils.sleep(retryWait);
                     response = await httpClient.delete(endpoint);
                   }
 
@@ -2134,7 +2174,7 @@ class FreeNASSshDriver extends ControllerZfsBaseDriver {
     // likely bad creds/url
     throw new GrpcError(
       grpc.status.UNKNOWN,
-      `FreeNAS error getting system version info: ${stringify({
+      `FreeNAS error getting system version info: ${GeneralUtils.stringify({
         errors: versionErrors,
         responses: versionResponses,
       })}`

src/driver/index.js

@@ -759,16 +759,21 @@ class CsiBaseDriver {
               }
 
               // create 'DB' entry
-              await iscsi.iscsiadm.createNodeDBEntry(
+              await GeneralUtils.retry(5, 2000, async () => {
-                iscsiConnection.iqn,
+                await iscsi.iscsiadm.createNodeDBEntry(
-                iscsiConnection.portal,
+                  iscsiConnection.iqn,
-                nodeDB
+                  iscsiConnection.portal,
-              );
+                  nodeDB
+                );
+              });
+
               // login
-              await iscsi.iscsiadm.login(
+              await GeneralUtils.retry(15, 2000, async () => {
-                iscsiConnection.iqn,
+                await iscsi.iscsiadm.login(
-                iscsiConnection.portal
+                  iscsiConnection.iqn,
-              );
+                  iscsiConnection.portal
+                );
+              });
 
               // get associated session
               let session = await iscsi.iscsiadm.getSession(

src/utils/general.js

@@ -1,3 +1,4 @@
+const _ = require("lodash");
 const axios = require("axios");
 const crypto = require("crypto");
 
@@ -177,6 +178,45 @@ function default_supported_file_filesystems() {
   return ["nfs", "cifs"];
 }
 
+async function retry(retries, retriesDelay, code, options = {}) {
+  let current_try = 0;
+  let maxwait = _.get(options, "maxwait");
+  let logerrors = _.get(options, "logerrors", false);
+  let retryCondition = options.retryCondition;
+  do {
+    current_try++;
+    try {
+      return await code();
+    } catch (err) {
+      if (current_try >= retries) {
+        throw err;
+      }
+      if (retryCondition) {
+        let retry = retryCondition(err);
+        if (!retry) {
+          console.log(`retry - failed condition, not trying again`);
+          throw err;
+        }
+      }
+      if (logerrors === true) {
+        console.log(`retry - err:`, err);
+      }
+    }
+    let sleep_time = retriesDelay;
+    if (_.get(options, "exponential", false) === true) {
+      sleep_time = retriesDelay * current_try;
+    }
+
+    if (maxwait) {
+      if (sleep_time > maxwait) {
+        sleep_time = maxwait;
+      }
+    }
+    console.log(`retry - waiting ${sleep_time}ms before trying again`);
+    await sleep(sleep_time);
+  } while (true);
+}
+
 module.exports.sleep = sleep;
 module.exports.md5 = md5;
 module.exports.crc32 = crc32;
@@ -189,3 +229,4 @@ module.exports.default_supported_block_filesystems =
   default_supported_block_filesystems;
 module.exports.default_supported_file_filesystems =
   default_supported_file_filesystems;
+module.exports.retry = retry;