Add an additional step in dataimport:

Wait for at least 10000 searchable docs in the meta-engine before exit
elastic · Apr 27, 2020 · e8a1f9a · e8a1f9a
1 parent 72a7f8b
commit e8a1f9a
Show file tree

Hide file tree

Showing 2 changed files with 43 additions and 14 deletions.
diff --git a/eas-kb-demo-dataimport/src/cli.js b/eas-kb-demo-dataimport/src/cli.js
@@ -67,20 +67,30 @@ const cliProgress = require('cli-progress');
             if (fileProgressBar) {
               fileProgressBar.update(0, {filename: inputFile.filename.padEnd(100)});
             }
-            const progressCallback = (batchSize) => {
+            const progressCallback = (importedDocs) => {
               if (fileProgressBar) {
-                fileProgressBar.increment(batchSize);
+                fileProgressBar.increment(importedDocs);
               }
             };
-            return client.importDocuments(inputFile.getEngineName(), docs, progressCallback).then(importStats => {
-              if (!fileProgressBar) {
-                const filename = inputFile.filename;
-                const engineName = inputFile.getEngineName();
-                console.info(`✔ Imported ${docs.length} documents from file ${filename} into engine ${engineName}.`);
-              }
-            });
+            return client.importDocuments(inputFile.getEngineName(), docs, progressCallback)
+              .then(({ importedDocs }) => {
+                if (!fileProgressBar) {
+                  const filename = inputFile.filename;
+                  const engineName = inputFile.getEngineName();
+                  console.info(`✔ Imported ${importedDocs} documents from file ${filename} into engine ${engineName}.`);
+                }
+                return Promise.resolve(importedDocs);
+              });
           });
-      })).then(() => progressBar.stop());
+      })).then(async(docCounts) => {
+        progressBar.stop();
+        const expectedDocCount = Math.min(10000, docCounts.reduce((acc, value) => acc + value, 0));
+        const { metaEngineName } = config;
+        while (await client.getSearchableDocCount(metaEngineName) < expectedDocCount) {
+          await new Promise(resolve => setTimeout(resolve, 100));
+        };
+        console.info(`✔ Meta engine ${metaEngineName} contains ${expectedDocCount} searchable documents.`);
+      });
     }).catch(reason => {
       throw reason;
     });

diff --git a/eas-kb-demo-dataimport/src/client/client.js b/eas-kb-demo-dataimport/src/client/client.js
@@ -127,20 +127,39 @@ class Client {
    * @param {*}        progressCallback A callback called each time for each imported chunk.
    */
   async importDocuments(engineName, documents, progressCallback) {
+    var importedDocs = 0;
     for (var i = 0; i < documents.length; i += this.batchSize) {
-      const cuurentBatch = documents.slice(i, i + this.batchSize);
+      const currentBatch = documents.slice(i, i + this.batchSize);
       try {
-        await this.client.indexDocuments(engineName, cuurentBatch);
+        const indexingResponse = await this.client.indexDocuments(engineName, currentBatch);
+
+        importedDocs += indexingResponse.filter(({errors}) => errors.length === 0).length;
+
         if (progressCallback) {
-          progressCallback(cuurentBatch.length);
+          progressCallback(currentBatch.length);
         }
 
       } catch ({ errorMessages: [message], ...error }) {
         return Promise.reject(`Error while importing documents in engine ${engineName} (${message})`);
       }
     }
 
-    return Promise.resolve({engine: engineName, docs: documents.length});
+    return Promise.resolve({ engine: engineName, importedDocs });
+  }
+
+  /**
+   * Return searchable doc count for the specified engine.
+   *
+   * @param {string} engineName Target engine name.
+   */
+  async getSearchableDocCount(engineName) {
+    try {
+      const searchResponse = await this.client.search(engineName, '', { page: { size: 1 } }).then();
+      const { meta: { page: { total_results: docCount } } } = searchResponse;
+      return Promise.resolve(docCount);
+    } catch ({ errorMessages: [message], ...error }) {
+      return Promise.reject(`Unable to retrieve searchable doc count for ${engineName} (${message})`);
+    }
   }
 }