system store - load from core - PR notes

Signed-off-by: Amit Prinz Setter <[email protected]>

Author: alphaprinz

config.js

@@ -250,7 +250,7 @@ config.INTERNAL_STORAGE_POOL_NAME = 'system-internal-storage-pool';
 config.ALLOW_BUCKET_CREATE_ON_INTERNAL = true;
 config.BUCKET_AUTOCONF_TIER2_ENABLED = false;
 config.SYSTEM_STORE_LOAD_CONCURRENCY = parseInt(process.env.SYSTEM_STORE_LOAD_CONCURRENCY, 10) || 5;
-
+config.SYSTEM_STORE_SOURCE = process.env.SYSTEM_STORE_SOURCE || "db";
 //////////////////////////
 // MD AGGREGATOR CONFIG //
 //////////////////////////

src/api/system_api.js

@@ -463,6 +463,12 @@ module.exports = {
 
         get_system_store: {
             method: 'GET',
+            reply: {
+                type: 'object',
+                properties: {
+                    // [RPC_BUFFERS].data
+                },
+            },
             auth: {
                 system: false
             }

src/endpoint/endpoint.js

@@ -171,7 +171,7 @@ async function main(options = {}) {
                 // Load a system store instance for the current process and register for changes.
                 // We do not wait for it in becasue the result or errors are not relevent at
                 // this point (and should not kill the process);
-                system_store.get_instance({source: system_store.SOURCE.CORE}).load();
+                system_store.get_instance().load();
                 // Register the process as an md_server.
                 await md_server.register_rpc();
             }

src/server/system_services/system_server.js

@@ -299,9 +299,13 @@ function get_system_status(req) {
 }
 
 async function get_system_store() {
-    return {
-        [RPC_BUFFERS]: {data: Buffer.from(JSON.stringify(await system_store.recent_db_data()))},
-    };
+    try {
+        return {
+            [RPC_BUFFERS]: {data: Buffer.from(JSON.stringify(await system_store.recent_db_data()))},
+        };
+    } catch (e) {
+        dbg.error("Failed getting system store", e);
+    }
 }
 
 async function _update_system_state(system_id, mode) {

src/server/system_services/system_store.js

@@ -357,7 +357,7 @@ class SystemStore extends EventEmitter {
         this.START_REFRESH_THRESHOLD = 10 * 60 * 1000;
         this.FORCE_REFRESH_THRESHOLD = 60 * 60 * 1000;
         this.SYSTEM_STORE_LOAD_CONCURRENCY = config.SYSTEM_STORE_LOAD_CONCURRENCY || 5;
-        this.source = (process.env.HOSTNAME && process.env.HOSTNAME.indexOf("endpoint") === -1) ? SOURCE.DB : SOURCE.CORE;
+        this.source = config.SYSTEM_STORE_SOURCE.toLocaleLowerCase() === 'core' ? SOURCE.CORE : SOURCE.DB;
         dbg.log0("system store source is", this.source);
         this._load_serial = new semaphore.Semaphore(1, { warning_timeout: this.START_REFRESH_THRESHOLD });
         for (const col of COLLECTIONS) {
@@ -431,11 +431,20 @@ class SystemStore extends EventEmitter {
                 const new_data = new SystemStoreData();
                 let millistamp = time_utils.millistamp();
                 await this._register_for_changes();
-                if (this.source === SOURCE.DB) {
+                let from_core_failure = false;
+
+                if (this.source === SOURCE.CORE) {
+                    try {
+                        this.data = new SystemStoreData();
+                        await this._read_new_data_from_core(this.data);
+                    } catch (e) {
+                        dbg.error("Failed to load system store from core. Will load from db.", e);
+                        from_core_failure = true;
+                    }
+                }
+
+                if (this.source === SOURCE.DB || from_core_failure) {
                     await this._read_new_data_from_db(new_data);
-                } else {
-                    this.data = new SystemStoreData();
-                    await this._read_new_data_from_core(this.data);
                 }
 
                 const secret = await os_utils.read_server_secret();
@@ -447,7 +456,7 @@ class SystemStore extends EventEmitter {
                         depth: 4
                     }));
                 }
-                if (this.source === SOURCE.DB) {
+                if (this.source === SOURCE.DB || from_core_failure) {
                     this.old_db_data = this._update_data_from_new(this.old_db_data || {}, new_data);
                     this.data = _.cloneDeep(this.old_db_data);
                 }