3.0.1-beta.1

Author: Igor Lins e Silva

chains/example.config.json

@@ -45,17 +45,20 @@
     "deltas": []
   },
   "scaling": {
-    "batch_size": 10000,
-    "queue_limit": 50000,
-    "readers": 1,
-    "ds_queues": 1,
-    "ds_threads": 1,
-    "ds_pool_size": 1,
-    "indexing_queues": 1,
-    "ad_idx_queues": 1,
-    "max_autoscale": 4,
+    "readers":            1,
+    "ds_queues":          1,
+    "ds_threads":         1,
+    "ds_pool_size":       1,
+    "indexing_queues":    1,
+    "ad_idx_queues":      1,
+    "max_autoscale":      4,
+    "batch_size":         5000,
+    "resume_trigger":     5000,
     "auto_scale_trigger": 20000,
-    "routing_mode": "heatmap"
+    "block_queue_limit":  10000,
+    "max_queue_limit":    100000,
+    "routing_mode":       "heatmap",
+    "polling_interval":   10000
   },
   "indexer": {
     "start_on": 0,
@@ -84,8 +87,12 @@
       "voters": true
     },
     "index_deltas": true,
-    "index_transfer_memo": true,
-    "index_all_deltas": true
+    "index_transfer_memo": false,
+    "index_all_deltas": true,
+    "deferred_trx": false,
+    "failed_trx": true,
+    "resource_limits": false,
+    "resource_usage": false
   },
   "prefetch": {
     "read": 50,

helpers/common_functions.ts

@@ -1,7 +1,14 @@
 import {ApiResponse, Client} from "@elastic/elasticsearch";
 import {Serialize} from "../addons/eosjs-native";
 
-const config = require(`../${process.env.CONFIG_JSON}`);
+let config;
+try {
+    config = require(`../${process.env.CONFIG_JSON}`);
+} catch (e) {
+    console.log(`Configuration not found: ${process.env.CONFIG_JSON}`);
+    process.exit(1);
+}
+
 const CHAIN = config.settings.chain;
 
 function getLastResult(results: ApiResponse) {

interfaces/hyperionConfig.ts

@@ -1,4 +1,8 @@
 export interface ScalingConfigs {
+    polling_interval: number;
+    resume_trigger: number;
+    max_queue_limit: number;
+    block_queue_limit: number;
     routing_mode: string;
     batch_size: number;
     queue_limit: number;
@@ -110,7 +114,11 @@ export interface HyperionConfig {
         },
         index_deltas: boolean,
         index_transfer_memo: boolean,
-        index_all_deltas: boolean
+        index_all_deltas: boolean,
+        deferred_trx: boolean,
+        failed_trx: boolean,
+        resource_usage: boolean,
+        resource_limits: boolean,
     };
 
     prefetch: {

modules/loader.ts

@@ -18,6 +18,10 @@ export class HyperionModuleLoader {
     constructor(private cm: ConfigurationModule) {
         this.conn = cm.connections;
         this.config = cm.config;
+        if (!this.conn.chains[this.config.settings.chain]) {
+            console.log('Chain  ' + this.config.settings.chain + ' not defined on connections.json!');
+            process.exit(0);
+        }
         this.chainID = this.conn.chains[this.config.settings.chain].chain_id;
         this.loadActionHandlers();
         this.loadParser().catch((err) => {

modules/master.ts

@@ -34,10 +34,10 @@ import * as cluster from "cluster";
 import {Worker} from "cluster";
 import {HyperionWorkerDef} from "../interfaces/hyperionWorkerDef";
 import {HyperionConfig} from "../interfaces/hyperionConfig";
-import moment = require("moment");
-import Timeout = NodeJS.Timeout;
 
 import {AsyncQueue, queue} from "async";
+import moment = require("moment");
+import Timeout = NodeJS.Timeout;
 
 export class HyperionMaster {
 
@@ -86,7 +86,6 @@ export class HyperionMaster {
     private lastProducer: string = null;
     private handoffCounter: number = 0;
     private missedRounds: object = {};
-    private blockMsgQueue: any[] = [];
 
     // IPC Messaging
     private totalMessages = 0;
@@ -134,6 +133,7 @@ export class HyperionMaster {
     private proposedSchedule: any;
     private wsRouterWorker: cluster.Worker;
     private liveBlockQueue: AsyncQueue<any>;
+    private readingPaused = false;
 
 
     constructor() {
@@ -1140,17 +1140,65 @@ export class HyperionMaster {
         }, 5000);
     }
 
-    private monitorIndexingQueues() {
-        const limit = this.conf.scaling.auto_scale_trigger;
-        const autoscaleConsumers = {};
-        setInterval(async () => {
-            const testedQueues = new Set();
-            for (const worker of this.workerMap) {
-                if (worker.worker_role === 'ingestor') {
-                    const queue = worker.queue;
-                    if (!testedQueues.has(queue)) {
-                        testedQueues.add(queue);
-                        const size = await this.manager.checkQueueSize(queue);
+    private async checkQueues(autoscaleConsumers, limit) {
+        const testedQueues = new Set();
+        for (const worker of this.workerMap) {
+            let queue = worker.queue;
+
+            if (worker.worker_role === 'ds_pool_worker') {
+                queue = `${this.chain}:ds_pool:${worker.local_id}`;
+            }
+
+            if (queue) {
+                if (!testedQueues.has(queue)) {
+                    testedQueues.add(queue);
+                    const size = await this.manager.checkQueueSize(queue);
+
+
+                    // pause readers if queues are above the max_limit
+                    if (size >= this.conf.scaling.max_queue_limit) {
+                        this.readingPaused = true;
+                        for (const worker of this.workerMap) {
+                            if (worker.worker_role === 'reader') {
+                                worker.wref.send({event: 'pause'});
+                            }
+                        }
+                    }
+
+                    // resume readers if the queues are below the trigger point
+                    if ((this.readingPaused && size <= this.conf.scaling.resume_trigger)) {
+                        this.readingPaused = false;
+                        for (const worker of this.workerMap) {
+                            if (worker.worker_role === 'reader') {
+                                worker.wref.send({event: 'pause'});
+                                worker.wref.send({
+                                    event: 'set_delay',
+                                    data: {
+                                        state: false,
+                                        delay: 0
+                                    }
+                                });
+                            }
+                        }
+                    }
+
+                    // apply block processing delay if 20% below max
+                    if (size >= this.conf.scaling.max_queue_limit * 0.8) {
+                        for (const worker of this.workerMap) {
+                            if (worker.worker_role === 'reader') {
+                                worker.wref.send({
+                                    event: 'set_delay',
+                                    data: {
+                                        state: true,
+                                        delay: 500
+                                    }
+                                });
+                            }
+                        }
+                    }
+
+
+                    if (worker.worker_role === 'ingestor') {
                         if (size > limit) {
                             if (!autoscaleConsumers[queue]) {
                                 autoscaleConsumers[queue] = 0;
@@ -1164,14 +1212,24 @@ export class HyperionMaster {
                                 });
                                 this.launchWorkers();
                                 autoscaleConsumers[queue]++;
-                            } else {
-                                // hLog(`WARN: Max consumer limit reached on ${queue}!`);
                             }
                         }
                     }
                 }
             }
-        }, 20000);
+        }
+    }
+
+    private monitorIndexingQueues() {
+        const limit = this.conf.scaling.auto_scale_trigger;
+        const autoscaleConsumers = {};
+        this.checkQueues(autoscaleConsumers, limit).catch(console.log);
+        if (!this.conf.scaling.polling_interval) {
+            this.conf.scaling.polling_interval = 20000;
+        }
+        setInterval(async () => {
+            await this.checkQueues(autoscaleConsumers, limit);
+        }, this.conf.scaling.polling_interval);
     }
 
     private onPm2Stop() {
@@ -1350,6 +1408,22 @@ export class HyperionMaster {
 
     async runMaster() {
 
+        // config checks
+        if (!this.conf.scaling.max_queue_limit) {
+            hLog(`scaling.max_queue_limit is not defined!`);
+            process.exit(1);
+        }
+
+        if (!this.conf.scaling.resume_trigger) {
+            hLog(`scaling.resume_trigger is not defined!`);
+            process.exit(1);
+        }
+
+        if (!this.conf.scaling.block_queue_limit) {
+            hLog(`scaling.block_queue_limit is not defined!`);
+            process.exit(1);
+        }
+
         this.printMode();
 
         // Preview mode - prints only the proposed worker map
@@ -1365,7 +1439,7 @@ export class HyperionMaster {
             this.printActiveProds();
         }
 
-        // ELasticsearch
+        // Elasticsearch
         this.client = this.manager.elasticsearchClient;
         try {
             const esInfo = await this.client.info();

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "hyperion-history",
-  "version": "3.0.0-beta.4",
+  "version": "3.0.1-beta.1",
   "description": "Scalable Full History API Solution for EOSIO based blockchains",
   "main": "launcher.js",
   "scripts": {
@@ -55,11 +55,11 @@
     "socket.io-redis": "5.2.0",
     "ws": "7.3.0",
     "yargs": "15.3.1",
-    "typescript": "3.9.2"
+    "typescript": "3.9.3"
   },
   "devDependencies": {
     "@types/ioredis": "4.16.2",
-    "@types/lodash": "4.14.151",
+    "@types/lodash": "4.14.152",
     "@types/amqplib": "0.5.13",
     "@types/async": "3.2.3",
     "@types/got": "9.6.11",

package.json

@@ -10,4 +10,4 @@ echo -e "\n-->> Starting $1..."
 echo -e "\n-->> Saving pm2 state..."
 (set -x; pm2 save)
 echo -e "\n-->> Reading $1 logs..."
-(set -x; pm2 logs --raw --lines 0 "$@")
+(set -x; pm2 logs --raw --lines 10 "$@")