Compare commits

...

42 Commits

Author SHA1 Message Date
Igor Lins e Silva 2c95cfbcf5 Update .gitignore 2021-09-13 04:16:05 -03:00
Igor Lins e Silva 8552dbf31e Update example.config.json 2021-09-13 04:14:46 -03:00
Igor Lins e Silva f545aad172 Update .gitignore 2021-09-13 04:14:36 -03:00
Igor Lins e Silva 5e68b54610 Update .gitignore 2021-09-13 04:14:16 -03:00
Igor Lins e Silva 1e79521ccc add 1 to the total indexed blocks to counter the missing block 1 on ship 2021-09-09 22:32:23 -03:00
Igor Lins e Silva 18ef675b88 cleanup 2021-09-09 03:27:14 -03:00
Igor Lins e Silva 0947f43b38 fix missing deltas on 2.1 2021-09-09 03:23:23 -03:00
Igor Lins e Silva 0d5d438378 fix heatmap mode 2021-09-09 02:54:13 -03:00
Igor Lins e Silva e678678fef assert queue to avoid race condition 2021-09-08 23:37:36 -03:00
Igor Lins e Silva 0eaba697eb Update get_abi_snapshot.ts 2021-09-08 23:18:56 -03:00
Igor Lins e Silva 80bd4adc8c 3.3.4-rc7
- include config disable_rate_limit
- minor package upgrades
- date parser fix
2021-09-06 10:23:58 -03:00
Igor Lins e Silva 81ae93b1bd Merge pull request #62 from eosrio/es-8-alpha
Es 8 alpha
2021-09-06 10:12:43 -03:00
Igor Lins e Silva 49d3b3434c Update elastic-routes.ts 2021-09-06 10:11:04 -03:00
Igor Lins e Silva 350d2107fe 3.3.4-rc7
validate date inputs before forwarding to ES
avoid duplicated checking of failed token balances
2021-09-06 10:04:24 -03:00
Igor Lins e Silva eda6655600 implement tx caching on v1 get_transaction 2021-08-26 01:39:39 -03:00
Igor Lins e Silva 8ed55f003d allow pre-deserialized block struct 2021-08-22 22:53:00 -03:00
Igor Lins e Silva db3c18fdf9 Update ecosystem_settings.js 2021-08-20 00:17:30 -03:00
Igor Lins e Silva 2d58c18048 Update ecosystem_settings.js 2021-08-20 00:16:32 -03:00
Igor Lins e Silva e21e3dd314 update packages 2021-08-17 00:38:51 -03:00
Igor Lins e Silva d854b339f9 cleanup 2021-08-17 00:32:55 -03:00
Igor Lins e Silva 3026ab9d25 remove type on bulk request 2021-08-17 00:31:22 -03:00
Igor Lins e Silva 9a7ee17508 Update elastic-routes.ts 2021-08-17 00:24:13 -03:00
Igor Lins e Silva 41dce730db Update elastic-routes.ts 2021-08-17 00:15:45 -03:00
Igor Lins e Silva 4bbc92d888 Update elastic-routes.ts 2021-08-17 00:14:00 -03:00
Igor Lins e Silva b08448812e re-introduce delta present as byte 2021-08-10 07:46:06 -03:00
Igor Lins e Silva 82f31f82ba merge tx_cache feature 2021-08-05 21:54:04 -03:00
Igor Lins e Silva 2a82e5842a v3.3.4-rc6
- update ws to v8
- add indexer.disable_delta_rm option
2021-08-05 20:50:15 -03:00
Igor Lins e Silva f27f5ca2a9 Update server.ts 2021-08-05 13:28:44 -03:00
Igor Lins e Silva cc87fab9cd update packages 3.3.4-rc6 2021-08-05 13:27:37 -03:00
Igor Lins e Silva 5242a757ac cleanup 2021-08-05 13:27:13 -03:00
Igor Lins e Silva d2a16e6059 fix get_transaction empty output + implement cached_lib + allow block_hint 2021-08-05 13:17:30 -03:00
Igor Lins e Silva 99c43af7bf docker updates 2021-07-31 04:45:52 -03:00
Igor Lins e Silva a8452d7906 Update fix-permissions.js 2021-07-27 06:03:39 -03:00
Igor Lins e Silva b8104a1f0c require node v16 2021-07-27 05:59:24 -03:00
Igor Lins e Silva d3b4e75513 Update .gitignore 2021-07-27 05:58:08 -03:00
Igor Lins e Silva 204bb63bd7 3.3.4-rc5 2021-07-21 07:40:02 -03:00
Igor Lins e Silva 034dccfbab Update index.d.ts 2021-07-20 06:17:15 -03:00
Igor Lins e Silva aafd4a3d63 Update routes.ts 2021-07-20 06:17:06 -03:00
Igor Lins e Silva cbe37827fc fix error handling on amqp 2021-07-12 14:23:40 -03:00
Igor Lins e Silva aa3487b238 Update example-connections.json 2021-07-12 00:15:49 -03:00
Igor Lins e Silva c090822d3b re-add frameMax patch 2021-07-12 00:14:30 -03:00
Igor Lins e Silva 0564efd4ac Update .gitignore 2021-07-09 07:22:50 -03:00
52 changed files with 5079 additions and 4778 deletions
+10 -31
View File
@@ -11,33 +11,8 @@ pids
*.seed
*.pid.lock
# Directory for instrumented libs generated by jscoverage/JSCover
lib-cov
# Coverage directory used by tools like istanbul
coverage
# nyc test coverage
.nyc_output
# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files)
.grunt
# Bower dependency directory (https://bower.io/)
bower_components
# node-waf configuration
.lock-wscript
# Compiled binary addons (https://nodejs.org/api/addons.html)
build/Release
# Dependency directories
node_modules/
jspm_packages/
# TypeScript v1 declaration files
typings/
# Optional npm cache directory
.npm
@@ -51,15 +26,9 @@ typings/
# Output of 'npm pack'
*.tgz
# Yarn Integrity file
.yarn-integrity
# dotenv environment variables file
.env
# next.js build output
.next
.idea
*.pem
ecosystem.config.js
@@ -106,4 +75,14 @@ modules/**/*.js.map
addons/**/*.js
addons/**/*.js.map
plugins/.state.json
plugins/repos
plugins/*.js
.gitmodules
docker/redis/data
docker/elasticsearch/data
docker/rabbitmq/data
docker/eosio/data
configuration_backups
+17 -7
View File
@@ -13,11 +13,21 @@ import fastifyRateLimit from 'fastify-rate-limit';
import fastify_eosjs from "./plugins/fastify-eosjs";
export function registerPlugins(server: Fastify.FastifyInstance<Server, IncomingMessage, ServerResponse>, params: any) {
server.register(fastifyElasticsearch, params.fastify_elasticsearch);
server.register(fastifySwagger, params.fastify_swagger);
server.register(fastifyCors);
server.register(fastifyFormbody);
server.register(fastifyRedis, params.fastify_redis);
server.register(fastifyRateLimit, params.fastify_rate_limit);
server.register(fastify_eosjs, params.fastify_eosjs);
server.register(fastifyElasticsearch, params.fastify_elasticsearch);
if (params.fastify_swagger) {
server.register(fastifySwagger, params.fastify_swagger);
}
server.register(fastifyCors);
server.register(fastifyFormbody);
server.register(fastifyRedis, params.fastify_redis);
if (params.fastify_rate_limit) {
server.register(fastifyRateLimit, params.fastify_rate_limit);
}
server.register(fastify_eosjs, params.fastify_eosjs);
}
+40 -1
View File
@@ -3,6 +3,7 @@ import {FastifyError, FastifyInstance, FastifyReply, FastifyRequest} from "fasti
import {createReadStream} from "fs";
import {addSharedSchemas, handleChainApiRedirect} from "./helpers/functions";
import autoLoad from 'fastify-autoload';
import got from "got";
function addRedirect(server: FastifyInstance, url: string, redirectTo: string) {
server.route({
@@ -28,6 +29,24 @@ function addRoute(server: FastifyInstance, handlersPath: string, prefix: string)
export function registerRoutes(server: FastifyInstance) {
// build internal map of routes
const routeSet = new Set<string>();
server.decorate('routeSet', routeSet);
const ignoreList = [
'/v2',
'/v2/history',
'/v2/state',
'/v1/chain/*',
'/v1/chain'
];
server.addHook('onRoute', opts => {
if (!ignoreList.includes(opts.url)) {
if (opts.url.startsWith('/v')) {
routeSet.add(opts.url);
}
}
});
// Register fastify api routes
addRoute(server, 'v2', '/v2');
addRoute(server, 'v2-history', '/v2/history');
@@ -54,8 +73,28 @@ export function registerRoutes(server: FastifyInstance) {
}
});
// /v1/node/get_supported_apis
server.route({
url: '/v1/node/get_supported_apis',
method: ["GET"],
schema: {
summary: "Get list of supported APIs",
tags: ["node"]
},
handler: async (request: FastifyRequest, reply: FastifyReply) => {
const data = await got.get(`${server.chain_api}/v1/node/get_supported_apis`).json() as any;
if (data.apis && data.apis.length > 0) {
const apiSet = new Set(server.routeSet);
data.apis.forEach((a) => apiSet.add(a));
reply.send({apis: [...apiSet]});
} else {
reply.send({apis: [...server.routeSet], error: 'nodeos did not send any data'});
}
}
});
server.addHook('onError', (request: FastifyRequest, reply: FastifyReply, error: FastifyError, done) => {
console.log(`[${request.headers['x-real-ip']}] ${request.method} ${request.url} failed with error: ${error.message}`);
console.log(`[${request.headers['x-real-ip'] || request.ip}] ${request.method} ${request.url} failed >> ${error.message}`);
done();
});
@@ -1,6 +1,6 @@
import {FastifyInstance, FastifyReply, FastifyRequest} from "fastify";
import {mergeActionMeta, timedQuery} from "../../../helpers/functions";
import * as flatstr from 'flatstr';
import flatstr from 'flatstr';
const terms = ["notified", "act.authorization.actor"];
const extendedActions = new Set(["transfer", "newaccount", "updateauth"]);
@@ -1,29 +1,19 @@
import {FastifyInstance, FastifyReply, FastifyRequest} from "fastify";
import {mergeActionMeta, timedQuery} from "../../../helpers/functions";
import {createHash} from "crypto";
import * as flatstr from 'flatstr';
import flatstr from 'flatstr';
import {GetInfoResult} from "eosjs/dist/eosjs-rpc-interfaces";
async function getTransaction(fastify: FastifyInstance, request: FastifyRequest) {
if (typeof request.body === 'string') {
request.body = JSON.parse(request.body)
}
const body: any = request.body;
const pResults = await Promise.all([fastify.eosjs.rpc.get_info(), fastify.elastic['search']({
"index": fastify.manager.chain + '-action-*',
"body": {
"query": {
"bool": {
must: [
{term: {"trx_id": body.id.toLowerCase()}}
]
}
},
"sort": {
"global_sequence": "asc"
}
}
})]);
const results = pResults[1];
const redis = fastify.redis;
const trxId = body.id.toLowerCase();
const conf = fastify.manager.config;
const cachedData = await redis.hgetall('trx_' + trxId);
const response: any = {
"id": body.id,
"trx": {
@@ -42,11 +32,89 @@ async function getTransaction(fastify: FastifyInstance, request: FastifyRequest)
},
"block_num": 0,
"block_time": "",
"last_irreversible_block": pResults[0].last_irreversible_block_num,
"last_irreversible_block": undefined,
"traces": []
};
const hits = results['body']['hits']['hits'];
let hits;
// build get_info request with caching
const $getInfo = new Promise<GetInfoResult>(resolve => {
const key = `${fastify.manager.chain}_get_info`;
fastify.redis.get(key).then(value => {
if (value) {
resolve(JSON.parse(value));
} else {
fastify.eosjs.rpc.get_info().then(value1 => {
fastify.redis.set(key, JSON.stringify(value1), 'EX', 6);
resolve(value1);
}).catch((reason) => {
console.log(reason);
response.error = 'failed to get last_irreversible_block_num'
resolve(null);
});
}
});
});
// reconstruct hits from cached data
if (cachedData && Object.keys(cachedData).length > 0) {
const gsArr = [];
for (let cachedDataKey in cachedData) {
gsArr.push(cachedData[cachedDataKey]);
}
gsArr.sort((a, b) => {
return a.global_sequence - b.global_sequence;
});
hits = gsArr.map(value => {
return {
_source: JSON.parse(value)
};
});
const promiseResults = await Promise.all([
redis.ttl('trx_' + trxId),
$getInfo
]);
response.cache_expires_in = promiseResults[0];
response.last_irreversible_block = promiseResults[1].last_irreversible_block_num;
}
// search on ES if cache is not present
if (!hits) {
const _size = conf.api.limits.get_trx_actions || 100;
const blockHint = parseInt(body.block_num_hint, 10);
let indexPattern = '';
if (blockHint) {
const idxPart = Math.ceil(blockHint / conf.settings.index_partition_size).toString().padStart(6, '0');
indexPattern = fastify.manager.chain + `-action-${conf.settings.index_version}-${idxPart}`;
} else {
indexPattern = fastify.manager.chain + '-action-*';
}
let pResults;
try {
// build search request
const $search = fastify.elastic.search({
index: indexPattern,
size: _size,
body: {
query: {bool: {must: [{term: {trx_id: trxId}}]}},
sort: {global_sequence: "asc"}
}
});
// execute in parallel
pResults = await Promise.all([$getInfo, $search]);
} catch (e) {
console.log(e.message);
if (e.meta.statusCode === 404) {
return response;
}
}
hits = pResults[1]['body']['hits']['hits'];
response.last_irreversible_block = pResults[0].last_irreversible_block_num;
}
if (hits.length > 0) {
const actions = hits;
@@ -9,7 +9,10 @@ export default function (fastify: FastifyInstance, opts: any, next) {
tags: ['history'],
body: {
type: ['object', 'string'],
properties: {id: {description: 'transaction id', type: 'string'}},
properties: {
id: {description: 'transaction id', type: 'string'},
block_num_hint: {description: 'block number hint', type: 'integer'},
},
required: ["id"]
}
});
@@ -12,7 +12,7 @@ async function checkTransaction(fastify: FastifyInstance, request: FastifyReques
status: jsonObj.status,
block_num: jsonObj.b,
root_action: jsonObj.a,
signatures: null,
signatures: [],
};
if (jsonObj.s && jsonObj.s.length > 0) {
response.signatures = jsonObj.s;
@@ -22,7 +22,7 @@ async function getAbiSnapshot(fastify: FastifyInstance, request: FastifyRequest)
}
const results = await fastify.elastic.search({
index: fastify.manager.chain + '-abi',
index: fastify.manager.chain + '-abi-*',
size: 1,
body: {
query: {bool: {must: mustArray}},
+202 -181
View File
@@ -1,215 +1,236 @@
import {primaryTerms, terms} from "./definitions";
export function addSortedBy(query, queryBody, sort_direction) {
if (query['sortedBy']) {
const opts = query['sortedBy'].split(":");
const sortedByObj = {};
sortedByObj[opts[0]] = opts[1];
queryBody['sort'] = sortedByObj;
} else {
queryBody['sort'] = {
"global_sequence": sort_direction
};
}
if (query['sortedBy']) {
const opts = query['sortedBy'].split(":");
const sortedByObj = {};
sortedByObj[opts[0]] = opts[1];
queryBody['sort'] = sortedByObj;
} else {
queryBody['sort'] = {
"global_sequence": sort_direction
};
}
}
export function processMultiVars(queryStruct, parts, field) {
const must = [];
const mustNot = [];
const must = [];
const mustNot = [];
parts.forEach(part => {
if (part.startsWith("!")) {
mustNot.push(part.replace("!", ""));
} else {
must.push(part);
}
});
parts.forEach(part => {
if (part.startsWith("!")) {
mustNot.push(part.replace("!", ""));
} else {
must.push(part);
}
});
if (must.length > 1) {
queryStruct.bool.must.push({
bool: {
should: must.map(elem => {
const _q = {};
_q[field] = elem;
return {term: _q}
})
}
});
} else if (must.length === 1) {
const mustQuery = {};
mustQuery[field] = must[0];
queryStruct.bool.must.push({term: mustQuery});
}
if (must.length > 1) {
queryStruct.bool.must.push({
bool: {
should: must.map(elem => {
const _q = {};
_q[field] = elem;
return {term: _q}
})
}
});
} else if (must.length === 1) {
const mustQuery = {};
mustQuery[field] = must[0];
queryStruct.bool.must.push({term: mustQuery});
}
if (mustNot.length > 1) {
queryStruct.bool.must_not.push({
bool: {
should: mustNot.map(elem => {
const _q = {};
_q[field] = elem;
return {term: _q}
})
}
});
} else if (mustNot.length === 1) {
const mustNotQuery = {};
mustNotQuery[field] = mustNot[0].replace("!", "");
queryStruct.bool.must_not.push({term: mustNotQuery});
}
if (mustNot.length > 1) {
queryStruct.bool.must_not.push({
bool: {
should: mustNot.map(elem => {
const _q = {};
_q[field] = elem;
return {term: _q}
})
}
});
} else if (mustNot.length === 1) {
const mustNotQuery = {};
mustNotQuery[field] = mustNot[0].replace("!", "");
queryStruct.bool.must_not.push({term: mustNotQuery});
}
}
function addRangeQuery(queryStruct, prop, pkey, query) {
const _termQuery = {};
const parts = query[prop].split("-");
_termQuery[pkey] = {
"gte": parts[0],
"lte": parts[1]
};
queryStruct.bool.must.push({range: _termQuery});
const _termQuery = {};
const parts = query[prop].split("-");
_termQuery[pkey] = {
"gte": parts[0],
"lte": parts[1]
};
queryStruct.bool.must.push({range: _termQuery});
}
export function applyTimeFilter(query, queryStruct) {
if (query['after'] || query['before']) {
let _lte = "now";
let _gte = "0";
if (query['before']) {
_lte = query['before'];
if (!_lte.endsWith("Z")) {
_lte += "Z";
}
}
if (query['after']) {
_gte = query['after'];
if (!_gte.endsWith("Z")) {
_gte += "Z";
}
}
if (!queryStruct.bool['filter']) {
queryStruct.bool['filter'] = [];
}
queryStruct.bool['filter'].push({
range: {
"@timestamp": {
"gte": _gte,
"lte": _lte
}
}
});
}
if (query['after'] || query['before']) {
let _lte = "now";
let _gte = "0";
if (query['before']) {
try {
_lte = new Date(query['before']).toISOString();
} catch (e) {
throw new Error(e.message + ' [before]');
}
}
if (query['after']) {
try {
_gte = new Date(query['after']).toISOString();
} catch (e) {
throw new Error(e.message + ' [after]');
}
}
if (!queryStruct.bool['filter']) {
queryStruct.bool['filter'] = [];
}
queryStruct.bool['filter'].push({
range: {
"@timestamp": {
"gte": _gte,
"lte": _lte
}
}
});
}
}
export function applyGenericFilters(query, queryStruct, allowedExtraParams: Set<string>) {
for (const prop in query) {
if (Object.prototype.hasOwnProperty.call(query, prop)) {
const pair = prop.split(".");
if (pair.length > 1 || primaryTerms.includes(pair[0])) {
let pkey;
if (pair.length > 1 && allowedExtraParams) {
pkey = allowedExtraParams.has(pair[0]) ? "@" + prop : prop;
} else {
pkey = prop;
}
if (query[prop].indexOf("-") !== -1) {
addRangeQuery(queryStruct, prop, pkey, query);
} else {
const _termQuery = {};
const parts = query[prop].split(",");
if (parts.length > 1) {
processMultiVars(queryStruct, parts, prop);
} else if (parts.length === 1) {
const andParts = parts[0].split(" ");
if (andParts.length > 1) {
andParts.forEach(value => {
const _q = {};
console.log(value);
_q[pkey] = value;
queryStruct.bool.must.push({term: _q});
});
} else {
if (parts[0].startsWith("!")) {
_termQuery[pkey] = parts[0].replace("!", "");
queryStruct.bool.must_not.push({term: _termQuery});
} else {
_termQuery[pkey] = parts[0];
queryStruct.bool.must.push({term: _termQuery});
}
}
}
}
}
}
}
for (const prop in query) {
if (Object.prototype.hasOwnProperty.call(query, prop)) {
const pair = prop.split(".");
if (pair.length > 1 || primaryTerms.includes(pair[0])) {
let pkey;
if (pair.length > 1 && allowedExtraParams) {
pkey = allowedExtraParams.has(pair[0]) ? "@" + prop : prop;
} else {
pkey = prop;
}
if (query[prop].indexOf("-") !== -1) {
addRangeQuery(queryStruct, prop, pkey, query);
} else {
const _qObj = {};
const parts = query[prop].split(",");
if (parts.length > 1) {
processMultiVars(queryStruct, parts, prop);
} else if (parts.length === 1) {
// @transfer.memo special case
if (pkey === '@transfer.memo') {
_qObj[pkey] = {
query: parts[0]
};
if (query.match_fuzziness) {
_qObj[pkey].fuzziness = query.match_fuzziness;
}
if (query.match_operator) {
_qObj[pkey].operator = query.match_operator;
}
queryStruct.bool.must.push({
match: _qObj
});
} else {
const andParts = parts[0].split(" ");
if (andParts.length > 1) {
andParts.forEach(value => {
const _q = {};
_q[pkey] = value;
queryStruct.bool.must.push({term: _q});
});
} else {
if (parts[0].startsWith("!")) {
_qObj[pkey] = parts[0].replace("!", "");
queryStruct.bool.must_not.push({term: _qObj});
} else {
_qObj[pkey] = parts[0];
queryStruct.bool.must.push({term: _qObj});
}
}
}
}
}
}
}
}
}
export function makeShouldArray(query) {
const should_array = [];
for (const entry of terms) {
const tObj = {term: {}};
tObj.term[entry] = query.account;
should_array.push(tObj);
}
return should_array;
const should_array = [];
for (const entry of terms) {
const tObj = {term: {}};
tObj.term[entry] = query.account;
should_array.push(tObj);
}
return should_array;
}
export function applyCodeActionFilters(query, queryStruct) {
let filterObj = [];
if (query.filter) {
for (const filter of query.filter.split(',')) {
if (filter !== '*:*') {
const _arr = [];
const parts = filter.split(':');
if (parts.length === 2) {
const [code, method] = parts;
if (code && code !== "*") {
_arr.push({'term': {'act.account': code}});
}
if (method && method !== "*") {
_arr.push({'term': {'act.name': method}});
}
}
if (_arr.length > 0) {
filterObj.push({bool: {must: _arr}});
}
}
}
if (filterObj.length > 0) {
queryStruct.bool['should'] = filterObj;
queryStruct.bool['minimum_should_match'] = 1;
}
}
let filterObj = [];
if (query.filter) {
for (const filter of query.filter.split(',')) {
if (filter !== '*:*') {
const _arr = [];
const parts = filter.split(':');
if (parts.length === 2) {
const [code, method] = parts;
if (code && code !== "*") {
_arr.push({'term': {'act.account': code}});
}
if (method && method !== "*") {
_arr.push({'term': {'act.name': method}});
}
}
if (_arr.length > 0) {
filterObj.push({bool: {must: _arr}});
}
}
}
if (filterObj.length > 0) {
queryStruct.bool['should'] = filterObj;
queryStruct.bool['minimum_should_match'] = 1;
}
}
}
export function getSkipLimit(query, max?: number) {
let skip, limit;
skip = parseInt(query.skip, 10);
if (skip < 0) {
throw new Error('invalid skip parameter');
}
limit = parseInt(query.limit, 10);
if (limit < 1) {
throw new Error('invalid limit parameter');
} else if (limit > max) {
throw new Error(`limit too big, maximum: ${max}`);
}
return {skip, limit};
let skip, limit;
skip = parseInt(query.skip, 10);
if (skip < 0) {
throw new Error('invalid skip parameter');
}
limit = parseInt(query.limit, 10);
if (limit < 1) {
throw new Error('invalid limit parameter');
} else if (limit > max) {
throw new Error(`limit too big, maximum: ${max}`);
}
return {skip, limit};
}
export function getSortDir(query) {
let sort_direction = 'desc';
if (query.sort) {
if (query.sort === 'asc' || query.sort === '1') {
sort_direction = 'asc';
} else if (query.sort === 'desc' || query.sort === '-1') {
sort_direction = 'desc'
} else {
throw new Error('invalid sort direction');
}
}
return sort_direction;
let sort_direction = 'desc';
if (query.sort) {
if (query.sort === 'asc' || query.sort === '1') {
sort_direction = 'asc';
} else if (query.sort === 'desc' || query.sort === '-1') {
sort_direction = 'desc'
} else {
throw new Error('invalid sort direction');
}
}
return sort_direction;
}
export function applyAccountFilters(query, queryStruct) {
if (query.account) {
queryStruct.bool.must.push({"bool": {should: makeShouldArray(query)}});
}
if (query.account) {
queryStruct.bool.must.push({"bool": {should: makeShouldArray(query)}});
}
}
+67 -63
View File
@@ -3,68 +3,72 @@ import {getDeltasHandler} from "./get_deltas";
import {addApiRoute, extendQueryStringSchema, extendResponseSchema, getRouteName} from "../../../helpers/functions";
export default function (fastify: FastifyInstance, opts: any, next) {
const schema: FastifySchema = {
description: 'get state deltas',
summary: 'get state deltas',
tags: ['history'],
querystring: extendQueryStringSchema({
"code": {
description: 'contract account',
type: 'string'
},
"scope": {
description: 'table scope',
type: 'string'
},
"table": {
description: 'table name',
type: 'string'
},
"payer": {
description: 'payer account',
type: 'string'
},
"after": {
description: 'filter after specified date (ISO8601)',
type: 'string'
},
"before": {
description: 'filter before specified date (ISO8601)',
type: 'string'
}
}),
response: extendResponseSchema({
"deltas": {
type: "array",
items: {
type: 'object',
properties: {
"timestamp": {type: 'string'},
"code": {type: 'string'},
"scope": {type: 'string'},
"table": {type: 'string'},
"primary_key": {type: 'string'},
"payer": {type: 'string'},
"present": {type: 'number'},
"block_num": {type: 'number'},
"block_id": {type: 'string'},
"data": {
type: 'object',
additionalProperties: true
}
},
additionalProperties: true
}
}
})
};
addApiRoute(
fastify,
'GET',
getRouteName(__filename),
getDeltasHandler,
schema
);
next();
const schema: FastifySchema = {
description: 'get state deltas',
summary: 'get state deltas',
tags: ['history'],
querystring: extendQueryStringSchema({
"code": {
description: 'contract account',
type: 'string'
},
"scope": {
description: 'table scope',
type: 'string'
},
"table": {
description: 'table name',
type: 'string'
},
"payer": {
description: 'payer account',
type: 'string'
},
"after": {
description: 'filter after specified date (ISO8601)',
type: 'string'
},
"before": {
description: 'filter before specified date (ISO8601)',
type: 'string'
},
"present": {
description: 'delta present flag',
type: 'number'
},
}),
response: extendResponseSchema({
"deltas": {
type: "array",
items: {
type: 'object',
properties: {
"timestamp": {type: 'string'},
"present": {type: 'number'},
"code": {type: 'string'},
"scope": {type: 'string'},
"table": {type: 'string'},
"primary_key": {type: 'string'},
"payer": {type: 'string'},
"block_num": {type: 'number'},
"block_id": {type: 'string'},
"data": {
type: 'object',
additionalProperties: true
}
},
additionalProperties: true
}
}
})
};
addApiRoute(
fastify,
'GET',
getRouteName(__filename),
getDeltasHandler,
schema
);
next();
}
@@ -1,130 +1,128 @@
import {FastifyInstance, FastifyReply, FastifyRequest} from "fastify";
import {mergeActionMeta, timedQuery} from "../../../helpers/functions";
import {GetInfoResult} from "eosjs/dist/eosjs-rpc-interfaces";
async function getTransaction(fastify: FastifyInstance, request: FastifyRequest) {
const redis = fastify.redis;
const query: any = request.query;
const trxId = query.id.toLowerCase();
const conf = fastify.manager.config;
const cachedData = await redis.hgetall('trx_' + trxId);
const response = {
query_time_ms: undefined,
executed: false,
cached: undefined,
cache_expires_in: undefined,
trx_id: query.id,
lib: undefined,
cached_lib: false,
actions: undefined,
generated: undefined,
error: undefined
};
let hits;
const query: any = request.query;
// build get_info request with caching
const $getInfo = new Promise<GetInfoResult>(resolve => {
const key = `${fastify.manager.chain}_get_info`;
fastify.redis.get(key).then(value => {
if (value) {
response.cached_lib = true;
resolve(JSON.parse(value));
} else {
fastify.eosjs.rpc.get_info().then(value1 => {
fastify.redis.set(key, JSON.stringify(value1), 'EX', 6);
response.cached_lib = false;
resolve(value1);
}).catch((reason) => {
console.log(reason);
response.error = 'failed to get last_irreversible_block_num'
resolve(null);
});
}
});
});
const _size = fastify.manager.config.api.limits.get_trx_actions || 100;
// reconstruct hits from cached data
if (cachedData && Object.keys(cachedData).length > 0) {
const gsArr = [];
for (let cachedDataKey in cachedData) {
gsArr.push(cachedData[cachedDataKey]);
}
gsArr.sort((a, b) => {
return a.global_sequence - b.global_sequence;
});
hits = gsArr.map(value => {
return {
_source: JSON.parse(value)
};
});
const promiseResults = await Promise.all([
redis.ttl('trx_' + trxId),
$getInfo
]);
response.cache_expires_in = promiseResults[0];
response.lib = promiseResults[1].last_irreversible_block_num;
}
let indexPattern = fastify.manager.chain + '-action-*';
if (query.hot_only) {
indexPattern = fastify.manager.chain + '-action';
}
// search on ES if cache is not present
if (!hits) {
const _size = conf.api.limits.get_trx_actions || 100;
const blockHint = parseInt(query.block_hint, 10);
let indexPattern = '';
if (blockHint) {
const idxPart = Math.ceil(blockHint / conf.settings.index_partition_size).toString().padStart(6, '0');
indexPattern = fastify.manager.chain + `-action-${conf.settings.index_version}-${idxPart}`;
} else {
indexPattern = fastify.manager.chain + '-action-*';
}
let pResults;
try {
const pResults = await Promise.all([
fastify.eosjs.rpc.get_info(),
fastify.elastic.search({
index: indexPattern,
size: _size,
body: {
query: {
bool: {
must: [
{term: {trx_id: query.id.toLowerCase()}}
]
}
},
sort: {
global_sequence: "asc"
}
}
}),
fastify.elastic.search({
index: fastify.manager.chain + '-gentrx-*',
size: _size,
body: {
query: {
bool: {
must: [
{term: {trx_id: query.id.toLowerCase()}}
]
}
}
}
})
]);
// build search request
const $search = fastify.elastic.search({
index: indexPattern,
size: _size,
body: {
query: {bool: {must: [{term: {trx_id: trxId}}]}},
sort: {global_sequence: "asc"}
}
});
const results = pResults[1];
const genTrxRes = pResults[2];
// execute in parallel
pResults = await Promise.all([$getInfo, $search]);
} catch (e) {
console.log(e.message);
if (e.meta.statusCode === 404) {
response.error = 'no data near block_hint'
return response;
}
}
hits = pResults[1]['body']['hits']['hits'];
response.lib = pResults[0].last_irreversible_block_num;
}
const response = {
"executed": false,
"hot_only": false,
"trx_id": query.id,
"lib": pResults[0].last_irreversible_block_num,
"actions": [],
"generated": undefined
};
if (query.hot_only) {
response.hot_only = true;
}
const hits = results['body']['hits']['hits'];
if (hits.length > 0) {
// const producers = {};
// for (let hit of hits) {
// if (hit._source.producer) {
// if (producers[hit._source.producer]) {
// producers[hit._source.producer]++;
// } else {
// producers[hit._source.producer] = 1;
// }
// }
// }
// let useBlocknumber;
// if (Object.keys(producers).length > 1) {
// // multiple producers of the same tx id, forked actions are present, attempt to cleanup
// let trueProd = '';
// let highestActCount = 0;
// for (const prod in producers) {
// if (producers.hasOwnProperty(prod)) {
// if(producers[prod] === highestActCount) {
// useBlocknumber = true;
// } else if (producers[prod] > highestActCount) {
// highestActCount = producers[prod];
// trueProd = prod;
// }
// }
// }
// }
let highestBlockNum = 0;
for (let action of hits) {
action = action._source;
if (action.block_num > highestBlockNum) {
highestBlockNum = action.block_num;
}
}
for (let action of hits) {
if (action.block_num === highestBlockNum) {
mergeActionMeta(action);
response.actions.push(action);
}
}
response.executed = true;
}
const hits2 = genTrxRes['body']['hits']['hits'];
if (hits2 && hits2.length > 0) {
if (hits2[0]._source['@timestamp']) {
hits2[0]._source['timestamp'] = hits2[0]._source['@timestamp'];
delete hits2[0]._source['@timestamp'];
}
response.generated = hits2[0]._source;
}
return response;
if (hits.length > 0) {
let highestBlockNum = 0;
for (let action of hits) {
if (action._source.block_num > highestBlockNum) {
highestBlockNum = action._source.block_num;
}
}
response.actions = [];
for (let action of hits) {
if (action._source.block_num === highestBlockNum) {
mergeActionMeta(action._source);
response.actions.push(action._source);
}
}
response.executed = true;
}
return response;
}
export function getTransactionHandler(fastify: FastifyInstance, route: string) {
return async (request: FastifyRequest, reply: FastifyReply) => {
reply.send(await timedQuery(getTransaction, fastify, request, route));
}
return async (request: FastifyRequest, reply: FastifyReply) => {
reply.send(await timedQuery(getTransaction, fastify, request, route));
}
}
@@ -10,9 +10,13 @@ export default function (fastify: FastifyInstance, opts: any, next) {
querystring: {
type: 'object',
properties: {
"id": {
id: {
description: 'transaction id',
type: 'string'
},
block_hint: {
description: 'block hint to speed up tx recovery',
type: 'integer'
}
},
required: ["id"]
+63 -56
View File
@@ -5,69 +5,76 @@ import {getSkipLimit} from "../../v2-history/get_actions/functions";
async function getTokens(fastify: FastifyInstance, request: FastifyRequest) {
const query: any = request.query;
const query: any = request.query;
const response = {'account': query.account, 'tokens': []};
const response = {'account': query.account, 'tokens': []};
const {skip, limit} = getSkipLimit(request.query);
const maxDocs = fastify.manager.config.api.limits.get_tokens ?? 100;
const {skip, limit} = getSkipLimit(request.query);
const maxDocs = fastify.manager.config.api.limits.get_tokens ?? 100;
const stateResult = await fastify.elastic.search({
"index": fastify.manager.chain + '-table-accounts-*',
"size": (limit > maxDocs ? maxDocs : limit) || 50,
"from": skip || 0,
"body": {
query: {
bool: {
filter: [{term: {"scope": query.account}}]
}
}
}
});
const stateResult = await fastify.elastic.search({
"index": fastify.manager.chain + '-table-accounts-*',
"size": (limit > maxDocs ? maxDocs : limit) || 50,
"from": skip || 0,
"body": {
query: {
bool: {
filter: [{term: {"scope": query.account}}]
}
}
}
});
for (const hit of stateResult.body.hits.hits) {
const data = hit._source;
if (typeof data.present !== "undefined" && data.present === 0) {
continue;
}
let precision;
const key = `${data.code}_${data.symbol}`;
if (!fastify.tokenCache) {
fastify.tokenCache = new Map<string, any>();
}
if (fastify.tokenCache.has(key)) {
precision = fastify.tokenCache.get(key).precision;
} else {
let token_data;
try {
token_data = await fastify.eosjs.rpc.get_currency_balance(data.code, query.account, data.symbol);
if (token_data.length > 0) {
const [amount, symbol] = token_data[0].split(" ");
const amount_arr = amount.split(".");
if (amount_arr.length === 2) {
precision = amount_arr[1].length;
fastify.tokenCache.set(key, {precision});
// console.log('Caching token precision -', key, precision);
}
}
} catch (e) {
console.log(`get_currency_balance error - contract:${data.code} - account:${query.account}`);
}
}
const testSet = new Set();
for (const hit of stateResult.body.hits.hits) {
const data = hit._source;
if (typeof data.present !== "undefined" && data.present === 0) {
continue;
}
let precision;
const key = `${data.code}_${data.symbol}`;
response.tokens.push({
symbol: data.symbol,
precision: precision,
amount: parseFloat(data.amount),
contract: data.code
});
}
if (testSet.has(key)) {
continue;
}
return response;
testSet.add(key);
if (!fastify.tokenCache) {
fastify.tokenCache = new Map<string, any>();
}
if (fastify.tokenCache.has(key)) {
precision = fastify.tokenCache.get(key).precision;
} else {
let token_data;
try {
token_data = await fastify.eosjs.rpc.get_currency_balance(data.code, query.account, data.symbol);
if (token_data.length > 0) {
const [amount, symbol] = token_data[0].split(" ");
const amount_arr = amount.split(".");
if (amount_arr.length === 2) {
precision = amount_arr[1].length;
fastify.tokenCache.set(key, {precision});
// console.log('Caching token precision -', key, precision);
}
}
} catch (e) {
console.log(`get_currency_balance error - contract:${data.code} - account:${query.account}`);
}
}
response.tokens.push({
symbol: data.symbol,
precision: precision,
amount: parseFloat(data.amount),
contract: data.code
});
}
return response;
}
export function getTokensHandler(fastify: FastifyInstance, route: string) {
return async (request: FastifyRequest, reply: FastifyReply) => {
reply.send(await timedQuery(getTokens, fastify, request, route));
}
return async (request: FastifyRequest, reply: FastifyReply) => {
reply.send(await timedQuery(getTokens, fastify, request, route));
}
}
+1 -1
View File
@@ -41,7 +41,7 @@ async function checkElastic(fastify: FastifyInstance) {
let indexedBlocks = await getLastIndexedBlockWithTotalBlocks(fastify.elastic, fastify.manager.chain);
const data = {
last_indexed_block: indexedBlocks[0],
total_indexed_blocks: indexedBlocks[1],
total_indexed_blocks: indexedBlocks[1] + 1,
active_shards: esStatus.body[0]['active_shards_percent']
};
let stat = 'OK';
+33 -26
View File
@@ -2,7 +2,7 @@ import {hLog} from "../helpers/common_functions";
import {ConfigurationModule} from "../modules/config";
import {ConnectionManager} from "../connections/manager.class";
import {HyperionConfig} from "../interfaces/hyperionConfig";
import * as Redis from 'ioredis';
import IORedis from 'ioredis';
import fastify from 'fastify'
import {registerPlugins} from "./plugins";
import {AddressInfo} from "net";
@@ -12,7 +12,7 @@ import {createWriteStream, existsSync, mkdirSync} from "fs";
import {SocketManager} from "./socketManager";
import {HyperionModuleLoader} from "../modules/loader";
import {extendedActions} from "./routes/v2-history/get_actions/definitions";
import {Socket, io} from "socket.io-client";
import {io, Socket} from "socket.io-client";
class HyperionApiServer {
@@ -91,38 +91,44 @@ class HyperionApiServer {
hLog(`Chain API URL: "${this.fastify.chain_api}" | Push API URL: "${this.fastify.push_api}"`);
const ioRedisClient = new Redis(this.manager.conn.redis);
const ioRedisClient = new IORedis(this.manager.conn.redis);
let rateLimiterWhitelist = ['127.0.0.1'];
const pluginParams = {
fastify_elasticsearch: {
client: this.manager.elasticsearchClient
},
fastify_redis: this.manager.conn.redis,
fastify_eosjs: this.manager,
} as any;
if (this.conf.api.rate_limit_allow && this.conf.api.rate_limit_allow.length > 0) {
const tempSet = new Set<string>(['127.0.0.1', ...this.conf.api.rate_limit_allow]);
rateLimiterWhitelist = [...tempSet];
if(!this.conf.api.disable_rate_limit) {
let rateLimiterWhitelist = ['127.0.0.1'];
if (this.conf.api.rate_limit_allow && this.conf.api.rate_limit_allow.length > 0) {
const tempSet = new Set<string>(['127.0.0.1', ...this.conf.api.rate_limit_allow]);
rateLimiterWhitelist = [...tempSet];
}
let rateLimiterRPM = 1000;
if (this.conf.api.rate_limit_rpm) {
rateLimiterRPM = this.conf.api.rate_limit_rpm;
}
pluginParams.fastify_rate_limit = {
max: rateLimiterRPM,
allowList: rateLimiterWhitelist,
timeWindow: '1 minute',
redis: ioRedisClient
}
}
let rateLimiterRPM = 1000;
if (this.conf.api.rate_limit_rpm) {
rateLimiterRPM = this.conf.api.rate_limit_rpm;
}
const api_rate_limit = {
max: rateLimiterRPM,
allowList: rateLimiterWhitelist,
timeWindow: '1 minute',
redis: ioRedisClient
};
if (this.conf.features.streaming.enable) {
this.activateStreaming();
}
registerPlugins(this.fastify, {
fastify_elasticsearch: {client: this.manager.elasticsearchClient},
fastify_swagger: generateOpenApiConfig(this.manager.config),
fastify_rate_limit: api_rate_limit,
fastify_redis: this.manager.conn.redis,
fastify_eosjs: this.manager,
});
const docsConfig = generateOpenApiConfig(this.manager.config);
if(docsConfig) {
pluginParams.fastify_swagger = docsConfig;
}
registerPlugins(this.fastify, pluginParams);
this.addGenericTypeParsing();
}
@@ -181,6 +187,7 @@ class HyperionApiServer {
registerRoutes(this.fastify);
// register documentation when ready
this.fastify.ready().then(async () => {
await this.fastify.swagger();
}, (err) => {
+2 -2
View File
@@ -3,7 +3,7 @@ import {Server, Socket} from 'socket.io';
import {createAdapter} from 'socket.io-redis';
import {io} from 'socket.io-client';
import {FastifyInstance} from "fastify";
import * as Redis from "ioredis";
import IORedis from "ioredis";
export interface StreamDeltasRequest {
code: string;
@@ -253,7 +253,7 @@ export class SocketManager {
transports: ['websocket', 'polling'],
});
const pubClient = new Redis(redisOptions);
const pubClient = new IORedis(redisOptions);
const subClient = pubClient.duplicate();
this.io.adapter(createAdapter({pubClient, subClient}));
+23 -16
View File
@@ -1,5 +1,7 @@
{
"api": {
"enabled": true,
"pm2_scaling": 1,
"chain_name": "EXAMPLE Chain",
"server_addr": "127.0.0.1",
"server_port": 7000,
@@ -23,14 +25,33 @@
"chain_api_error_log": false,
"custom_core_token": "",
"enable_export_action": false,
"disable_rate_limit": false,
"rate_limit_rpm": 1000,
"rate_limit_allow": []
"rate_limit_allow": [],
"disable_tx_cache": false,
"tx_cache_expiration_sec": 3600
},
"indexer": {
"enabled": true,
"start_on": 0,
"stop_on": 0,
"rewrite": false,
"purge_queues": false,
"live_reader": false,
"live_only_mode": false,
"abi_scan_mode": true,
"fetch_block": true,
"fetch_traces": true,
"disable_reading": false,
"disable_indexing": false,
"process_deltas": true,
"disable_delta_rm": true
},
"settings": {
"preview": false,
"chain": "eos",
"eosio_alias": "eosio",
"parser": "1.8",
"parser": "2.1",
"auto_stop": 0,
"index_version": "v1",
"debug": false,
@@ -74,20 +95,6 @@
"routing_mode": "round_robin",
"polling_interval": 10000
},
"indexer": {
"start_on": 0,
"stop_on": 0,
"rewrite": false,
"purge_queues": false,
"live_reader": false,
"live_only_mode": false,
"abi_scan_mode": true,
"fetch_block": true,
"fetch_traces": true,
"disable_reading": false,
"disable_indexing": false,
"process_deltas": true
},
"features": {
"streaming": {
"enable": false,
+13 -7
View File
@@ -17,10 +17,14 @@ export async function createConnection(config): Promise<Connection> {
}
export function getAmpqUrl(config): string {
let frameMaxValue = '0x10000';
if (config.frameMax) {
frameMaxValue = config.frameMax;
}
const u = encodeURIComponent(config.user);
const p = encodeURIComponent(config.pass);
const v = encodeURIComponent(config.vhost)
return `amqp://${u}:${p}@${config.host}/${v}`;
return `amqp://${u}:${p}@${config.host}/${v}?frameMax=${frameMaxValue}`;
}
@@ -66,7 +70,7 @@ export async function amqpConnect(onReconnect, config, onClose) {
export async function checkQueueSize(q_name, config) {
try {
const v = encodeURIComponent(config.vhost);
const apiUrl = `http://${config.api}/api/queues/${v}/${encodeURIComponent(q_name)}`;
const apiUrl = `${config.protocol}://${config.api}/api/queues/${v}/${encodeURIComponent(q_name)}`;
const opts = {
username: config.user,
password: config.pass
@@ -74,11 +78,13 @@ export async function checkQueueSize(q_name, config) {
const data = await got.get(apiUrl, opts).json() as any;
return data.messages;
} catch (e) {
hLog('[WARNING] Checking queue size failed, HTTP API is not ready!');
if (e instanceof HTTPError) {
hLog(e.response.body);
} else {
hLog(JSON.stringify(e.response.body, null, 2));
hLog(`[WARNING] Checking queue size failed! - ${e.message}`);
if (e.response && e.response.body) {
if (e instanceof HTTPError) {
hLog(e.response.body);
} else {
hLog(JSON.stringify(e.response.body, null, 2));
}
}
return 0;
}
+3
View File
@@ -25,6 +25,9 @@ export class ConnectionManager {
constructor(private cm: ConfigurationModule) {
this.config = cm.config;
this.conn = cm.connections;
if (!this.conn.amqp.protocol) {
this.conn.amqp.protocol = 'http';
}
this.chain = this.config.settings.chain;
this.esIngestClients = [];
this.prepareESClient();
+40 -39
View File
@@ -1,48 +1,49 @@
import {debugLog, hLog} from "../helpers/common_functions";
const WebSocket = require('ws');
import WebSocket from 'ws';
export class StateHistorySocket {
private ws;
private readonly shipUrl;
private readonly max_payload_mb;
private ws;
private readonly shipUrl;
private readonly max_payload_mb;
constructor(ship_url, max_payload_mb) {
this.shipUrl = ship_url;
if (max_payload_mb) {
this.max_payload_mb = max_payload_mb;
} else {
this.max_payload_mb = 256;
}
}
constructor(ship_url, max_payload_mb) {
this.shipUrl = ship_url;
if (max_payload_mb) {
this.max_payload_mb = max_payload_mb;
} else {
this.max_payload_mb = 256;
}
}
connect(onMessage, onDisconnect, onError, onConnected) {
debugLog(`Connecting to ${this.shipUrl}...`);
this.ws = new WebSocket(this.shipUrl, null, {
perMessageDeflate: false,
maxPayload: this.max_payload_mb * 1024 * 1024,
});
this.ws.on('open', () => {
hLog('Websocket connected!');
if (onConnected) {
onConnected();
}
});
this.ws.on('message', onMessage);
this.ws.on('close', () => {
hLog('Websocket disconnected!');
onDisconnect();
});
this.ws.on('error', (err) => {
hLog(`${this.shipUrl} :: ${err.message}`);
});
}
connect(onMessage, onDisconnect, onError, onConnected) {
close() {
this.ws.close();
}
debugLog(`Connecting to ${this.shipUrl}...`);
this.ws = new WebSocket(this.shipUrl, {
perMessageDeflate: false,
maxPayload: this.max_payload_mb * 1024 * 1024,
});
this.ws.on('open', () => {
hLog('Websocket connected!');
if (onConnected) {
onConnected();
}
});
this.ws.on('message', (data) => onMessage(data));
this.ws.on('close', () => {
hLog('Websocket disconnected!');
onDisconnect();
});
this.ws.on('error', (err) => {
hLog(`${this.shipUrl} :: ${err.message}`);
});
}
send(payload) {
this.ws.send(payload);
}
close() {
this.ws.close();
}
send(payload) {
this.ws.send(payload);
}
}
+232 -228
View File
@@ -1,260 +1,264 @@
const AbiDefinitions = {
version: "eosio::abi/1.1",
structs: [
version: 'eosio::abi/1.1',
structs: [
{
name: 'extensions_entry',
base: '',
fields: [
{
name: "extensions_entry",
base: "",
fields: [
{
name: "tag",
type: "uint16"
},
{
name: "value",
type: "bytes"
}
]
name: 'tag',
type: 'uint16',
},
{
name: "type_def",
base: "",
fields: [
{
name: "new_type_name",
type: "string"
},
{
name: "type",
type: "string"
}
]
name: 'value',
type: 'bytes',
},
],
},
{
name: 'type_def',
base: '',
fields: [
{
name: 'new_type_name',
type: 'string',
},
{
name: "field_def",
base: "",
fields: [
{
name: "name",
type: "string"
},
{
name: "type",
type: "string"
}
]
name: 'type',
type: 'string',
},
],
},
{
name: 'field_def',
base: '',
fields: [
{
name: 'name',
type: 'string',
},
{
name: "struct_def",
base: "",
fields: [
{
name: "name",
type: "string"
},
{
name: "base",
type: "string"
},
{
name: "fields",
type: "field_def[]"
}
]
name: 'type',
type: 'string',
},
],
},
{
name: 'struct_def',
base: '',
fields: [
{
name: 'name',
type: 'string',
},
{
name: "action_def",
base: "",
fields: [
{
name: "name",
type: "name"
},
{
name: "type",
type: "string"
},
{
name: "ricardian_contract",
type: "string"
}
]
name: 'base',
type: 'string',
},
{
name: "table_def",
base: "",
fields: [
{
name: "name",
type: "name"
},
{
name: "index_type",
type: "string"
},
{
name: "key_names",
type: "string[]"
},
{
name: "key_types",
type: "string[]"
},
{
name: "type",
type: "string"
}
]
name: 'fields',
type: 'field_def[]',
},
],
},
{
name: 'action_def',
base: '',
fields: [
{
name: 'name',
type: 'name',
},
{
name: "clause_pair",
base: "",
fields: [
{
name: "id",
type: "string"
},
{
name: "body",
type: "string"
}
]
name: 'type',
type: 'string',
},
{
name: "error_message",
base: "",
fields: [
{
name: "error_code",
type: "uint64"
},
{
name: "error_msg",
type: "string"
}
]
name: 'ricardian_contract',
type: 'string',
},
],
},
{
name: 'table_def',
base: '',
fields: [
{
name: 'name',
type: 'name',
},
{
name: "variant_def",
base: "",
fields: [
{
name: "name",
type: "string"
},
{
name: "types",
type: "string[]"
}
]
name: 'index_type',
type: 'string',
},
{
name: "abi_def",
base: "",
fields: [
{
name: "version",
type: "string"
},
{
name: "types",
type: "type_def[]"
},
{
name: "structs",
type: "struct_def[]"
},
{
name: "actions",
type: "action_def[]"
},
{
name: "tables",
type: "table_def[]"
},
{
name: "ricardian_clauses",
type: "clause_pair[]"
},
{
name: "error_messages",
type: "error_message[]"
},
{
name: "abi_extensions",
type: "extensions_entry[]"
},
{
name: "variants",
type: "variant_def[]$"
}
]
}
]
name: 'key_names',
type: 'string[]',
},
{
name: 'key_types',
type: 'string[]',
},
{
name: 'type',
type: 'string',
},
],
},
{
name: 'clause_pair',
base: '',
fields: [
{
name: 'id',
type: 'string',
},
{
name: 'body',
type: 'string',
},
],
},
{
name: 'error_message',
base: '',
fields: [
{
name: 'error_code',
type: 'uint64',
},
{
name: 'error_msg',
type: 'string',
},
],
},
{
name: 'variant_def',
base: '',
fields: [
{
name: 'name',
type: 'string',
},
{
name: 'types',
type: 'string[]',
},
],
},
{
name: 'abi_def',
base: '',
fields: [
{
name: 'version',
type: 'string',
},
{
name: 'types',
type: 'type_def[]',
},
{
name: 'structs',
type: 'struct_def[]',
},
{
name: 'actions',
type: 'action_def[]',
},
{
name: 'tables',
type: 'table_def[]',
},
{
name: 'ricardian_clauses',
type: 'clause_pair[]',
},
{
name: 'error_messages',
type: 'error_message[]',
},
{
name: 'abi_extensions',
type: 'extensions_entry[]',
},
{
name: 'variants',
type: 'variant_def[]$',
},
],
},
],
};
const RexAbi = {
version: "eosio::abi/1.1",
types: [],
structs: [
version: 'eosio::abi/1.1',
types: [],
structs: [
{
name: 'buyresult',
base: '',
fields: [
{
name: "buyresult",
base: "",
fields: [{
name: "rex_received",
type: "asset"
}
]
name: 'rex_received',
type: 'asset',
},
],
}, {
name: 'orderresult',
base: '',
fields: [
{
name: 'owner',
type: 'name',
}, {
name: "orderresult",
base: "",
fields: [{
name: "owner",
type: "name"
}, {
name: "proceeds",
type: "asset"
}
]
}, {
name: "rentresult",
base: "",
fields: [{
name: "rented_tokens",
type: "asset"
}
]
name: 'proceeds',
type: 'asset',
},
],
}, {
name: 'rentresult',
base: '',
fields: [
{
name: "sellresult",
base: "",
fields: [{
name: "proceeds",
type: "asset"
}]
}
],
actions: [
{
name: "buyresult",
type: "buyresult",
ricardian_contract: ""
name: 'rented_tokens',
type: 'asset',
},
],
},
{
name: 'sellresult',
base: '',
fields: [
{
name: "orderresult",
type: "orderresult",
ricardian_contract: ""
},
{
name: "rentresult",
type: "rentresult",
ricardian_contract: ""
},
{
name: "sellresult",
type: "sellresult",
ricardian_contract: ""
}
]
name: 'proceeds',
type: 'asset',
}],
},
],
actions: [
{
name: 'buyresult',
type: 'buyresult',
ricardian_contract: '',
},
{
name: 'orderresult',
type: 'orderresult',
ricardian_contract: '',
},
{
name: 'rentresult',
type: 'rentresult',
ricardian_contract: '',
},
{
name: 'sellresult',
type: 'sellresult',
ricardian_contract: '',
},
],
};
module.exports = {AbiDefinitions, RexAbi};
+12 -3
View File
@@ -4,17 +4,25 @@ const idx_js_heap = 4096;
// max js heap in MB for each api process
const api_js_heap = 1024;
function interpreterArgs(heap) {
const arr = ['--max-old-space-size=' + heap, '--trace-deprecation', '--trace-warnings'];
if (process.env.INSPECT) {
arr.push('--inspect');
}
return arr;
}
function addIndexer(chainName) {
return {
script: './launcher.js',
name: chainName + '-indexer',
namespace: chainName,
interpreter: 'node',
interpreter_args: ['--max-old-space-size=' + idx_js_heap, '--trace-deprecation', '--trace-warnings'],
interpreter_args: interpreterArgs(idx_js_heap),
autorestart: false,
kill_timeout: 3600,
watch: false,
time: true, // include timestamps in pm2 logs
time: true,
env: {
CONFIG_JSON: 'chains/' + chainName + '.config.json',
TRACE_LOGS: 'false',
@@ -27,13 +35,14 @@ function addApiServer(chainName, threads) {
script: './api/server.js',
name: chainName + '-api',
namespace: chainName,
node_args: ['--max-old-space-size=' + api_js_heap, '--trace-deprecation', '--trace-warnings'],
node_args: interpreterArgs(api_js_heap),
exec_mode: 'cluster',
merge_logs: true,
instances: threads,
autorestart: true,
exp_backoff_restart_delay: 100,
watch: false,
time: true,
env: {
CONFIG_JSON: 'chains/' + chainName + '.config.json',
},
+1
View File
@@ -221,6 +221,7 @@ export const delta = {
// base fields
"@timestamp": {"type": "date"},
"present": {"type": "byte"},
"ds_error": {"type": "boolean"},
"block_id": {"type": "keyword"},
"block_num": {"type": "long"},
@@ -0,0 +1,5 @@
cluster.name: docker-cluster
network.host: 0.0.0.0
bootstrap.memory_lock: true
discovery.type: single-node
node.name: es01
@@ -0,0 +1,2 @@
-Xms8g
-Xmx8g
+5 -11
View File
@@ -1,11 +1,5 @@
FROM ubuntu:18.04
RUN apt-get update && apt-get upgrade -y && apt-get autoremove && apt-get install -y wget netcat
RUN wget -nv https://github.com/eosio/eos/releases/download/v2.0.5/eosio_2.0.5-1-ubuntu-18.04_amd64.deb
RUN apt-get install -y ./eosio_2.0.5-1-ubuntu-18.04_amd64.deb
RUN useradd -m -s /bin/bash eosio
USER eosio
EXPOSE 8080
FROM ubuntu:20.04
LABEL version="2.1.0" description="EOSIO 2.1.0"
RUN apt-get update && apt-get install -y wget
RUN wget -nv https://github.com/EOSIO/eos/releases/download/v2.1.0/eosio_2.1.0-1-ubuntu-20.04_amd64.deb
RUN apt-get install -y ./eosio_2.1.0-1-ubuntu-20.04_amd64.deb && rm -f ./eosio_2.1.0-1-ubuntu-20.04_amd64.deb
+7 -21
View File
@@ -1,28 +1,14 @@
# Enable block production, even if the chain is stale. (eosio::producer_plugin)
### PRODUCER PLUGIN ###
enable-stale-production = true
# ID of producer controlled by this node (eosio::producer_plugin)
producer-name = eosio
# print contract's output to console (eosio::chain_plugin)
contracts-console = true
# The local IP and port to listen for incoming http connections; set blank to disable. (eosio::http_plugin)
http-server-address = 0.0.0.0:8888
# If set to false, then any incoming "Host" header is considered valid (eosio::http_plugin)
http-validate-host = false
# enable trace history (eosio::state_history_plugin)
### STATE HISTORY PLUGIN ###
plugin = eosio::state_history_plugin
state-history-endpoint = 0.0.0.0:8080
trace-history = true
# enable chain state history (eosio::state_history_plugin)
chain-state-history = true
# the endpoint upon which to listen for incoming connections. Caution: only expose this port to your internal network. (eosio::state_history_plugin)
state-history-endpoint = 0.0.0.0:8080
# Plugin(s) to enable, may be specified multiple times
plugin = eosio::producer_api_plugin
### CHAIN API PLUGIN ###
plugin = eosio::chain_api_plugin
plugin = eosio::state_history_plugin
http-server-address = 0.0.0.0:8888
http-validate-host = false
+5 -14
View File
@@ -1,15 +1,6 @@
FROM ubuntu:18.04
RUN apt-get update && apt-get upgrade -y && apt-get autoremove && apt-get install -y build-essential git curl netcat
RUN curl -sL https://deb.nodesource.com/setup_14.x | bash -
RUN apt-get install -y nodejs && npm install pm2 -g && git clone https://github.com/eosrio/hyperion-history-api.git
RUN useradd -m -s /bin/bash hyperion && chown -R hyperion:hyperion /hyperion-history-api
USER hyperion
FROM node:16.5
RUN npm install pm2 -g
RUN git clone https://github.com/eosrio/hyperion-history-api.git
WORKDIR /hyperion-history-api
RUN npm install
EXPOSE 7001
RUN git checkout dev-3.3
RUN npm install --production
-5
View File
@@ -1,5 +0,0 @@
# Ignore everything in this directory
*
# Except this file
!.gitignore
+4
View File
@@ -0,0 +1,4 @@
default_vhost = hyperion
default_user = hyperion
default_pass = hyp123456
cluster_name = hyp-docker
-111
View File
@@ -1,111 +0,0 @@
version: '3.8'
services:
redis:
container_name: redis
image: redis:5.0.9-alpine
restart: on-failure
networks:
- hyperion
rabbitmq:
container_name: rabbitmq
image: rabbitmq:3.8.3-management
restart: on-failure
environment:
- RABBITMQ_DEFAULT_USER=username
- RABBITMQ_DEFAULT_PASS=password
- RABBITMQ_DEFAULT_VHOST=/hyperion
volumes:
- ./rabbitmq/data:/var/lib/rabbitmq
ports:
- 15672:15672
networks:
- hyperion
elasticsearch:
container_name: elasticsearch
image: docker.elastic.co/elasticsearch/elasticsearch:7.7.1
restart: on-failure
environment:
- discovery.type=single-node
- cluster.name=es-cluster
- node.name=es01
- bootstrap.memory_lock=true
- xpack.security.enabled=true
- "ES_JAVA_OPTS=-Xms2g -Xmx2g"
- ELASTIC_USERNAME=elastic
- ELASTIC_PASSWORD=password
volumes:
- ./elasticsearch/data:/usr/share/elasticsearch/data
ports:
- 9200:9200
networks:
- hyperion
kibana:
container_name: kibana
image: docker.elastic.co/kibana/kibana:7.7.1
restart: on-failure
environment:
- ELASTICSEARCH_HOSTS=http://elasticsearch:9200
- ELASTICSEARCH_USERNAME=elastic
- ELASTICSEARCH_PASSWORD=password
ports:
- 5601:5601
networks:
- hyperion
depends_on:
- elasticsearch
eosio-node:
container_name: eosio-node
image: eosrio/hyperion:eosio-2.0.5
volumes:
- ./eosio/data/:/home/eosio/data/
- ./eosio/config/:/home/eosio/config/
- ./scripts/:/home/eosio/scripts/
ports:
- 8888:8888
networks:
- hyperion
command: bash -c "/home/eosio/scripts/run-nodeos.sh ${SCRIPT:-false} ${SNAPSHOT:-""}"
hyperion-indexer:
container_name: hyperion-indexer
image: eosrio/hyperion:hyperion-3.1.0-beta.3
restart: on-failure
depends_on:
- elasticsearch
- redis
- rabbitmq
- eosio-node
volumes:
- ./hyperion/config/connections.json:/hyperion-history-api/connections.json
- ./hyperion/config/ecosystem.config.js:/hyperion-history-api/ecosystem.config.js
- ./hyperion/config/chains/:/hyperion-history-api/chains/
- ./scripts/:/home/hyperion/scripts/
networks:
- hyperion
command: bash -c "/home/hyperion/scripts/run-hyperion.sh ${SCRIPT:-false} eos-indexer"
hyperion-api:
container_name: hyperion-api
image: eosrio/hyperion:hyperion-3.0
restart: on-failure
ports:
- 7000:7000
depends_on:
- hyperion-indexer
volumes:
- ./hyperion/config/connections.json:/hyperion-history-api/connections.json
- ./hyperion/config/ecosystem.config.js:/hyperion-history-api/ecosystem.config.js
- ./hyperion/config/chains/:/hyperion-history-api/chains/
- ./scripts/:/home/hyperion/scripts/
networks:
- hyperion
command: bash -c "/home/hyperion/scripts/run-hyperion.sh ${SCRIPT:-false} eos-api"
networks:
hyperion:
driver: bridge
+6 -2
View File
@@ -2,14 +2,18 @@
"amqp": {
"host": "127.0.0.1:5672",
"api": "127.0.0.1:15672",
"protocol": "http",
"user": "username",
"pass": "password",
"vhost": "hyperion"
"vhost": "hyperion",
"frameMax": "0x10000"
},
"elasticsearch": {
"protocol": "http",
"host": "127.0.0.1:9200",
"ingest_nodes": ["127.0.0.1:9200"],
"ingest_nodes": [
"127.0.0.1:9200"
],
"user": "elastic",
"pass": "password"
},
+456 -452
View File
@@ -1,6 +1,6 @@
import {Channel, Message} from "amqplib/callback_api";
import {ConnectionManager} from "../connections/manager.class";
import * as _ from "lodash";
import _ from "lodash";
import {hLog} from "./common_functions";
import {createHash} from "crypto";
@@ -9,534 +9,538 @@ type routerFunction = (blockNum: number) => string;
type MMap = Map<string, any>;
function makeScriptedOp(id, body) {
return [
{update: {_id: id, retry_on_conflict: 3}},
{script: {id: "updateByBlock", params: body}, scripted_upsert: true, upsert: {}}
];
return [
{update: {_id: id, retry_on_conflict: 3}},
{script: {id: "updateByBlock", params: body}, scripted_upsert: true, upsert: {}}
];
}
function makeDelOp(id) {
return [
{delete: {_id: id}}
];
return [
{delete: {_id: id}}
];
}
function flatMap(payloads: Message[], builder) {
return _(payloads).map((payload: Message) => {
const body = JSON.parse(payload.content.toString());
return builder(payload, body);
}).flatten()['value']();
return _(payloads).map((payload: Message) => {
const body = JSON.parse(payload.content.toString());
return builder(payload, body);
}).flatten()['value']();
}
function buildAbiBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = body['block'] + body['account'];
messageMap.set(id, _.omit(payload, ['content']));
return [{index: {_id: id}}, body];
});
return flatMap(payloads, (payload, body) => {
const id = body['block'] + body['account'];
messageMap.set(id, _.omit(payload, ['content']));
return [{index: {_id: id}}, body];
});
}
function buildActionBulk(payloads, messageMap: MMap, maxBlockCb: MaxBlockCb, routerFunc: routerFunction, indexName: string) {
return flatMap(payloads, (payload, body) => {
const id = body['global_sequence'];
messageMap.set(id, _.omit(payload, ['content']));
return [{
index: {
_id: id,
_index: `${indexName}-${routerFunc(body.block_num)}`
}
}, body];
});
return flatMap(payloads, (payload, body) => {
const id = body['global_sequence'];
messageMap.set(id, _.omit(payload, ['content']));
return [{
index: {
_id: id,
_index: `${indexName}-${routerFunc(body.block_num)}`
}
}, body];
});
}
function buildBlockBulk(payloads, messageMap: MMap, maxBlockCb: MaxBlockCb, routerFunc: routerFunction, indexName: string) {
return flatMap(payloads, (payload, body) => {
const id = body['block_num'];
messageMap.set(id, _.omit(payload, ['content']));
return [{
index: {_id: id}
}, body];
});
return flatMap(payloads, (payload, body) => {
const id = body['block_num'];
messageMap.set(id, _.omit(payload, ['content']));
return [{
index: {_id: id}
}, body];
});
}
function buildDeltaBulk(payloads, messageMap: MMap, maxBlockCb: MaxBlockCb, routerFunc: routerFunction, indexName: string) {
let maxBlock = 0;
const flat_map = flatMap(payloads, (payload, b) => {
delete b.present;
if (maxBlock < b.block_num) {
maxBlock = b.block_num;
}
const id = `${b.block_num}-${b.code}-${b.scope}-${b.table}-${b.primary_key}`;
messageMap.set(id, _.omit(payload, ['content']));
return [{
index: {
_id: id,
_index: `${indexName}-${routerFunc(b.block_num)}`
}
}, b];
});
maxBlockCb(maxBlock);
return flat_map;
let maxBlock = 0;
const flat_map = flatMap(payloads, (payload, b) => {
if (maxBlock < b.block_num) {
maxBlock = b.block_num;
}
const id = `${b.block_num}-${b.code}-${b.scope}-${b.table}-${b.primary_key}`;
messageMap.set(id, _.omit(payload, ['content']));
return [{
index: {
_id: id,
_index: `${indexName}-${routerFunc(b.block_num)}`
}
}, b];
});
maxBlockCb(maxBlock);
return flat_map;
}
function buildDynamicTableBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
messageMap.set(payload.id, _.omit(payload, ['content']));
if (payload.present === 0) {
return makeDelOp(payload.id);
} else {
return makeScriptedOp(payload.id, body);
}
});
return flatMap(payloads, (payload, body) => {
messageMap.set(payload.id, _.omit(payload, ['content']));
if (payload.present === 0) {
return makeDelOp(payload.id);
} else {
return makeScriptedOp(payload.id, body);
}
});
}
function buildTableProposalsBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = `${body.proposer}-${body.proposal_name}-${body.primary_key}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
return flatMap(payloads, (payload, body) => {
const id = `${body.proposer}-${body.proposal_name}-${body.primary_key}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
}
function buildTableAccountsBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = `${body.code}-${body.scope}-${body.symbol}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
return flatMap(payloads, (payload, body) => {
const id = `${body.code}-${body.scope}-${body.symbol}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
}
function buildTableVotersBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = `${body.voter}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
return flatMap(payloads, (payload, body) => {
const id = `${body.voter}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
}
function buildLinkBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = `${body.account}-${body.permission}-${body.code}-${body.action}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
return flatMap(payloads, (payload, body) => {
const id = `${body.account}-${body.permission}-${body.code}-${body.action}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
}
function buildPermBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = `${body.owner}-${body.name}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
return flatMap(payloads, (payload, body) => {
const id = `${body.owner}-${body.name}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
}
function buildResLimitBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = `${body.owner}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
return flatMap(payloads, (payload, body) => {
const id = `${body.owner}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
}
function buildResUsageBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = `${body.owner}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
return flatMap(payloads, (payload, body) => {
const id = `${body.owner}`;
messageMap.set(id, _.omit(payload, ['content']));
return makeScriptedOp(id, body);
});
}
function buildGenTrxBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const hash = createHash('sha256')
.update(body.sender + body.sender_id + body.payer)
.digest()
.toString("hex");
const id = `${body.block_num}-${hash}`;
messageMap.set(id, _.omit(payload, ['content']));
return [{index: {_id: id}}, body];
});
return flatMap(payloads, (payload, body) => {
const hash = createHash('sha256')
.update(body.sender + body.sender_id + body.payer)
.digest()
.toString("hex");
const id = `${body.block_num}-${hash}`;
messageMap.set(id, _.omit(payload, ['content']));
return [{index: {_id: id}}, body];
});
}
function buildTrxErrBulk(payloads, messageMap: MMap) {
return flatMap(payloads, (payload, body) => {
const id = body.trx_id.toLowerCase();
delete body.trx_id;
messageMap.set(id, _.omit(payload, ['content']));
return [{index: {_id: id}}, body];
});
return flatMap(payloads, (payload, body) => {
const id = body.trx_id.toLowerCase();
delete body.trx_id;
messageMap.set(id, _.omit(payload, ['content']));
return [{index: {_id: id}}, body];
});
}
const generatorsMap = {
permission: {
index_name: 'perm',
func: buildPermBulk
},
permission_link: {
index_name: 'link',
func: buildLinkBulk
},
resource_limits: {
index_name: 'reslimits',
func: buildResLimitBulk
},
resource_usage: {
index_name: 'userres',
func: buildResUsageBulk
},
generated_transaction: {
index_name: 'gentrx',
func: buildGenTrxBulk
},
trx_error: {
index_name: 'trxerr',
func: buildTrxErrBulk
},
permission: {
index_name: 'perm',
func: buildPermBulk
},
permission_link: {
index_name: 'link',
func: buildLinkBulk
},
resource_limits: {
index_name: 'reslimits',
func: buildResLimitBulk
},
resource_usage: {
index_name: 'userres',
func: buildResUsageBulk
},
generated_transaction: {
index_name: 'gentrx',
func: buildGenTrxBulk
},
trx_error: {
index_name: 'trxerr',
func: buildTrxErrBulk
},
};
interface IndexDist {
index: string;
first_block: number;
last_block: number;
index: string;
first_block: number;
last_block: number;
}
export class ElasticRoutes {
public routes: any;
cm: ConnectionManager;
chain: string;
ingestNodeCounters = {};
distributionMap: IndexDist[];
public routes: any;
cm: ConnectionManager;
chain: string;
ingestNodeCounters = {};
distributionMap: IndexDist[];
constructor(connectionManager: ConnectionManager, distributionMap: IndexDist[]) {
this.distributionMap = distributionMap;
this.routes = {generic: this.handleGenericRoute.bind(this)};
this.cm = connectionManager;
this.chain = this.cm.chain;
this.registerRoutes();
this.resetCounters();
}
constructor(connectionManager: ConnectionManager, distributionMap: IndexDist[]) {
this.distributionMap = distributionMap;
this.routes = {generic: this.handleGenericRoute.bind(this)};
this.cm = connectionManager;
this.chain = this.cm.chain;
this.registerRoutes();
this.resetCounters();
}
createGenericBuilder(collection, channel, counter, index_name, method) {
return new Promise((resolve) => {
const messageMap = new Map();
let maxBlockNum;
this.bulkAction({
index: index_name,
body: method(collection, messageMap, (maxBlock) => {
maxBlockNum = maxBlock;
})
}).then((resp: any) => {
if (resp.body.errors) {
this.ackOrNack(resp, messageMap, channel);
} else {
if (maxBlockNum > 0) {
ElasticRoutes.reportMaxBlock(maxBlockNum, index_name);
}
channel.ackAll();
}
resolve(messageMap.size);
}).catch((err) => {
try {
channel.nackAll();
hLog('NackAll', err);
} finally {
resolve(0);
}
});
})
}
createGenericBuilder(collection, channel, counter, index_name, method) {
return new Promise((resolve) => {
const messageMap = new Map();
let maxBlockNum;
this.bulkAction({
index: index_name,
body: method(collection, messageMap, (maxBlock) => {
maxBlockNum = maxBlock;
})
}).then((resp: any) => {
if (resp.body.errors) {
this.ackOrNack(resp, messageMap, channel);
} else {
if (maxBlockNum > 0) {
ElasticRoutes.reportMaxBlock(maxBlockNum, index_name);
}
channel.ackAll();
}
resolve(messageMap.size);
}).catch((err) => {
try {
channel.nackAll();
hLog('NackAll', err);
} finally {
resolve(0);
}
});
})
}
async handleGenericRoute(payload: Message[], ch: Channel, cb: (indexed_size: number) => void): Promise<void> {
const coll = {};
for (const message of payload) {
const type = message.properties.headers.type;
if (!coll[type]) {
coll[type] = [];
}
coll[type].push(message);
}
async handleGenericRoute(payload: Message[], ch: Channel, cb: (indexed_size: number) => void): Promise<void> {
const coll = {};
for (const message of payload) {
const type = message.properties.headers.type;
if (!coll[type]) {
coll[type] = [];
}
coll[type].push(message);
}
const queue = [];
const v = this.cm.config.settings.index_version;
let counter = 0;
const queue = [];
const v = this.cm.config.settings.index_version;
let counter = 0;
Object.keys(coll).forEach(value => {
if (generatorsMap[value]) {
const indexName = `${this.chain}-${generatorsMap[value].index_name}-${v}`;
queue.push(this.createGenericBuilder(coll[value], ch, counter, indexName, generatorsMap[value].func));
}
});
Object.keys(coll).forEach(value => {
if (generatorsMap[value]) {
const indexName = `${this.chain}-${generatorsMap[value].index_name}-${v}`;
queue.push(this.createGenericBuilder(coll[value], ch, counter, indexName, generatorsMap[value].func));
}
});
await Promise.all(queue);
cb(counter);
}
await Promise.all(queue);
cb(counter);
}
resetCounters() {
this.cm.ingestClients.forEach((val, idx) => {
this.ingestNodeCounters[idx] = {
status: true,
docs: 0
};
});
}
resetCounters() {
this.cm.ingestClients.forEach((val, idx) => {
this.ingestNodeCounters[idx] = {
status: true,
docs: 0
};
});
}
ackOrNack(resp, messageMap, channel: Channel) {
for (const item of resp.body.items) {
let id, itemBody;
if (item['index']) {
id = item.index._id;
itemBody = item.index;
} else if (item['update']) {
id = item.update._id;
itemBody = item.update;
} else if (item['delete']) {
id = item.delete._id;
itemBody = item.delete;
} else {
console.log(item);
throw new Error('FATAL ERROR - CANNOT EXTRACT ID');
}
const message = messageMap.get(id);
const status = itemBody.status;
if (message) {
switch (status) {
case 200: {
channel.ack(message);
break;
}
case 201: {
channel.ack(message);
break;
}
case 404: {
channel.ack(message);
break;
}
case 409: {
console.log(item);
channel.nack(message);
break;
}
default: {
console.log(item, message.fields.deliveryTag);
console.info(`nack id: ${id} - status: ${status}`);
channel.nack(message);
}
}
} else {
console.log(item);
throw new Error('Message not found');
}
}
}
ackOrNack(resp, messageMap, channel: Channel) {
for (const item of resp.body.items) {
let id, itemBody;
if (item['index']) {
id = item.index._id;
itemBody = item.index;
} else if (item['update']) {
id = item.update._id;
itemBody = item.update;
} else if (item['delete']) {
id = item.delete._id;
itemBody = item.delete;
} else {
console.log(item);
throw new Error('FATAL ERROR - CANNOT EXTRACT ID');
}
const message = messageMap.get(id);
const status = itemBody.status;
if (message) {
switch (status) {
case 200: {
channel.ack(message);
break;
}
case 201: {
channel.ack(message);
break;
}
case 404: {
channel.ack(message);
break;
}
case 409: {
console.log(item);
channel.nack(message);
break;
}
default: {
console.log(item, message.fields.deliveryTag);
console.info(`nack id: ${id} - status: ${status}`);
channel.nack(message);
}
}
} else {
console.log(item);
throw new Error('Message not found');
}
}
}
onResponse(resp, messageMap, callback, payloads, channel: Channel, index_name: string, maxBlockNum: number) {
if (resp.errors) {
this.ackOrNack(resp, messageMap, channel);
if (maxBlockNum > 0) {
ElasticRoutes.reportMaxBlock(maxBlockNum, index_name);
}
} else {
channel.ackAll();
}
callback(messageMap.size);
}
onResponse(resp, messageMap, callback, payloads, channel: Channel, index_name: string, maxBlockNum: number) {
if (resp.errors) {
this.ackOrNack(resp, messageMap, channel);
if (maxBlockNum > 0) {
ElasticRoutes.reportMaxBlock(maxBlockNum, index_name);
}
} else {
channel.ackAll();
}
callback(messageMap.size);
}
onError(err, channel: Channel, callback) {
try {
channel.nackAll();
hLog('NackAll', err);
} finally {
callback();
}
}
onError(err, channel: Channel, callback) {
try {
channel.nackAll();
if (err.meta.body) {
hLog('NackAll:', JSON.stringify(err.meta.body.error, null, 2));
} else {
hLog('NackAll:', err);
}
} finally {
callback();
}
}
bulkAction(bulkData) {
let minIdx = 0;
if (this.cm.ingestClients.length > 1) {
let min;
this.cm.ingestClients.forEach((val, idx) => {
if (!min) {
min = this.ingestNodeCounters[idx].docs;
} else {
if (this.ingestNodeCounters[idx].docs < min) {
min = this.ingestNodeCounters[idx].docs;
minIdx = idx;
}
}
});
}
this.ingestNodeCounters[minIdx].docs += bulkData.body.length;
if (this.ingestNodeCounters[minIdx].docs > 1000) {
this.resetCounters();
}
return this.cm.ingestClients[minIdx]['bulk'](bulkData);
}
bulkAction(bulkData) {
let minIdx = 0;
if (this.cm.ingestClients.length > 1) {
let min;
this.cm.ingestClients.forEach((val, idx) => {
if (!min) {
min = this.ingestNodeCounters[idx].docs;
} else {
if (this.ingestNodeCounters[idx].docs < min) {
min = this.ingestNodeCounters[idx].docs;
minIdx = idx;
}
}
});
}
this.ingestNodeCounters[minIdx].docs += bulkData.body.length;
if (this.ingestNodeCounters[minIdx].docs > 10000) {
this.resetCounters();
}
// print full request
// console.log(JSON.stringify(bulkData));
return this.cm.ingestClients[minIdx]['bulk'](bulkData);
}
getIndexPartition(blockNum: number): string {
return Math.ceil(blockNum / this.cm.config.settings.index_partition_size).toString().padStart(6, '0');
}
getIndexPartition(blockNum: number): string {
return Math.ceil(blockNum / this.cm.config.settings.index_partition_size).toString().padStart(6, '0');
}
getIndexNameByBlock(block_num) {
if (!this.distributionMap) {
return null;
}
for (let i = this.distributionMap.length - 1; i >= 0; i--) {
const test = this.distributionMap[i].first_block <= block_num && this.distributionMap[i].last_block >= block_num;
if (test) {
return this.distributionMap[i].index;
}
}
return null;
}
getIndexNameByBlock(block_num) {
if (!this.distributionMap) {
return null;
}
for (let i = this.distributionMap.length - 1; i >= 0; i--) {
const test = this.distributionMap[i].first_block <= block_num && this.distributionMap[i].last_block >= block_num;
if (test) {
return this.distributionMap[i].index;
}
}
return null;
}
addToIndexMap(map, idx, payload) {
if (idx) {
if (!map[idx]) {
map[idx] = [];
}
map[idx].push(payload);
}
}
addToIndexMap(map, idx, payload) {
if (idx) {
if (!map[idx]) {
map[idx] = [];
}
map[idx].push(payload);
}
}
private routeFactory(indexName: string, bulkGenerator, routerFunction) {
return async (payloads, channel, cb) => {
private routeFactory(indexName: string, bulkGenerator, routerFunction) {
return async (payloads, channel, cb) => {
let _index = `${this.chain}-${indexName}-${this.cm.config.settings.index_version}`;
let _index = `${this.chain}-${indexName}-${this.cm.config.settings.index_version}`;
// write directly to index
const messageMap = new Map();
let maxBlockNum;
this.bulkAction({
index: _index,
type: '_doc',
body: bulkGenerator(payloads, messageMap, (maxBlock) => {
maxBlockNum = maxBlock;
}, routerFunction, _index)
}).then(resp => {
this.onResponse(resp, messageMap, cb, payloads, channel, _index, maxBlockNum);
}).catch(err => {
this.onError(err, channel, cb);
});
// write directly to index
const messageMap = new Map();
let maxBlockNum;
this.bulkAction({
index: _index,
body: bulkGenerator(payloads, messageMap, (maxBlock) => {
maxBlockNum = maxBlock;
}, routerFunction, _index)
}).then(resp => {
this.onResponse(resp, messageMap, cb, payloads, channel, _index, maxBlockNum);
}).catch(err => {
this.onError(err, channel, cb);
});
// if (this.cm.config.indexer.rewrite) {
//
// // // write to remapped indices
// // let payloadBlock = null;
// // const indexMap = {};
// // let pIdx = null;
// // if (indexName === 'action' || indexName === 'delta') {
// // for (const payload of payloads) {
// // const blk = payload.properties.headers?.block_num;
// // if (!payloadBlock) {
// // payloadBlock = blk;
// // pIdx = this.getIndexNameByBlock(blk);
// // console.log(pIdx);
// // this.addToIndexMap(indexMap, pIdx, payload);
// // } else {
// // const _idx = payloadBlock === blk ? pIdx : this.getIndexNameByBlock(blk);
// // console.log(_idx);
// // this.addToIndexMap(indexMap, _idx, payload);
// // }
// // }
// // }
// //
// // // if no index was mapped to that range use the default alias
// // if (Object.keys(indexMap).length === 0) {
// // indexMap[_index] = payloads;
// // }
// //
// // const queue = [];
// // let counter = 0;
// //
// // for (const idxKey in indexMap) {
// // queue.push(this.createGenericBuilder(indexMap[idxKey], channel, counter, idxKey, bulkGenerator));
// // }
// //
// // const results = await Promise.all(queue);
// // results.forEach(value => counter += value);
// // cb(counter);
//
// } else {
// // write directly to index
// const messageMap = new Map();
// let maxBlockNum;
// this.bulkAction({
// index: _index,
// type: '_doc',
// body: bulkGenerator(payloads, messageMap, (maxBlock) => {
// maxBlockNum = maxBlock;
// }, routerFunction, _index)
// }).then(resp => {
// this.onResponse(resp, messageMap, cb, payloads, channel, _index, maxBlockNum);
// }).catch(err => {
// this.onError(err, channel, cb);
// });
// }
};
}
// if (this.cm.config.indexer.rewrite) {
//
// // // write to remapped indices
// // let payloadBlock = null;
// // const indexMap = {};
// // let pIdx = null;
// // if (indexName === 'action' || indexName === 'delta') {
// // for (const payload of payloads) {
// // const blk = payload.properties.headers?.block_num;
// // if (!payloadBlock) {
// // payloadBlock = blk;
// // pIdx = this.getIndexNameByBlock(blk);
// // console.log(pIdx);
// // this.addToIndexMap(indexMap, pIdx, payload);
// // } else {
// // const _idx = payloadBlock === blk ? pIdx : this.getIndexNameByBlock(blk);
// // console.log(_idx);
// // this.addToIndexMap(indexMap, _idx, payload);
// // }
// // }
// // }
// //
// // // if no index was mapped to that range use the default alias
// // if (Object.keys(indexMap).length === 0) {
// // indexMap[_index] = payloads;
// // }
// //
// // const queue = [];
// // let counter = 0;
// //
// // for (const idxKey in indexMap) {
// // queue.push(this.createGenericBuilder(indexMap[idxKey], channel, counter, idxKey, bulkGenerator));
// // }
// //
// // const results = await Promise.all(queue);
// // results.forEach(value => counter += value);
// // cb(counter);
//
// } else {
// // write directly to index
// const messageMap = new Map();
// let maxBlockNum;
// this.bulkAction({
// index: _index,
// type: '_doc',
// body: bulkGenerator(payloads, messageMap, (maxBlock) => {
// maxBlockNum = maxBlock;
// }, routerFunction, _index)
// }).then(resp => {
// this.onResponse(resp, messageMap, cb, payloads, channel, _index, maxBlockNum);
// }).catch(err => {
// this.onError(err, channel, cb);
// });
// }
};
}
private addRoute(indexType: string, bulkGenerator, routerFunction) {
this.routes[indexType] = this.routeFactory(indexType, bulkGenerator, routerFunction);
}
private addRoute(indexType: string, bulkGenerator, routerFunction) {
this.routes[indexType] = this.routeFactory(indexType, bulkGenerator, routerFunction);
}
private registerRoutes() {
this.registerDynamicTableRoute();
const partitionRouter = this.getIndexPartition.bind(this);
this.addRoute('abi', buildAbiBulk, null);
this.addRoute('action', buildActionBulk, partitionRouter);
this.addRoute('block', buildBlockBulk, partitionRouter);
this.addRoute('delta', buildDeltaBulk, partitionRouter);
this.addRoute('table-voters', buildTableVotersBulk, null);
this.addRoute('table-accounts', buildTableAccountsBulk, null);
this.addRoute('table-proposals', buildTableProposalsBulk, null);
// this.addRoute('dynamic-table', buildDynamicTableBulk);
}
private registerRoutes() {
this.registerDynamicTableRoute();
const partitionRouter = this.getIndexPartition.bind(this);
this.addRoute('abi', buildAbiBulk, null);
this.addRoute('action', buildActionBulk, partitionRouter);
this.addRoute('block', buildBlockBulk, partitionRouter);
this.addRoute('delta', buildDeltaBulk, partitionRouter);
this.addRoute('table-voters', buildTableVotersBulk, null);
this.addRoute('table-accounts', buildTableAccountsBulk, null);
this.addRoute('table-proposals', buildTableProposalsBulk, null);
// this.addRoute('dynamic-table', buildDynamicTableBulk);
}
private registerDynamicTableRoute() {
this.routes['dynamic-table'] = async (payloads, channel, cb) => {
const contractMap: Map<string, any[]> = new Map();
let counter = 0;
for (const payload of payloads) {
const headers = payload?.properties?.headers;
const item = {
id: headers.id,
block_num: headers.block_num,
present: headers.present,
content: payload.content,
fields: payload.fields,
properties: payload.properties
};
if (contractMap.has(headers.code)) {
contractMap.get(headers.code).push(item);
} else {
contractMap.set(headers.code, [item]);
}
}
const processingQueue = [];
for (const entry of contractMap.entries()) {
processingQueue.push(
this.createGenericBuilder(
entry[1],
channel,
counter,
`${this.chain}-dt-${entry[0]}`,
buildDynamicTableBulk
)
);
}
const results = await Promise.all(processingQueue);
results.forEach(value => counter += value);
cb(counter);
};
}
private registerDynamicTableRoute() {
this.routes['dynamic-table'] = async (payloads, channel, cb) => {
const contractMap: Map<string, any[]> = new Map();
let counter = 0;
for (const payload of payloads) {
const headers = payload?.properties?.headers;
const item = {
id: headers.id,
block_num: headers.block_num,
present: headers.present,
content: payload.content,
fields: payload.fields,
properties: payload.properties
};
if (contractMap.has(headers.code)) {
contractMap.get(headers.code).push(item);
} else {
contractMap.set(headers.code, [item]);
}
}
const processingQueue = [];
for (const entry of contractMap.entries()) {
processingQueue.push(
this.createGenericBuilder(
entry[1],
channel,
counter,
`${this.chain}-dt-${entry[0]}`,
buildDynamicTableBulk
)
);
}
const results = await Promise.all(processingQueue);
results.forEach(value => counter += value);
cb(counter);
};
}
private static reportMaxBlock(maxBlockNum: number, index_name: string) {
process.send({
event: 'ingestor_block_report',
index: index_name,
proc: process.env.worker_role,
block_num: maxBlockNum
});
}
private static reportMaxBlock(maxBlockNum: number, index_name: string) {
process.send({
event: 'ingestor_block_report',
index: index_name,
proc: process.env.worker_role,
block_num: maxBlockNum
});
}
}
-51
View File
@@ -1,51 +0,0 @@
import * as cluster from "cluster";
import {ConfigurationModule} from "./modules/config";
import {HyperionWorker} from "./workers/hyperionWorker";
import {hLog} from "./helpers/common_functions";
interface WorkerEnv {
worker_role: string;
worker_id: string;
}
const hyperionWorkers = {
ds_pool_worker: 'ds-pool',
reader: 'state-reader',
deserializer: 'deserializer',
continuous_reader: 'state-reader',
ingestor: 'indexer',
router: 'ws-router',
delta_updater: 'delta-updater'
};
export async function launch() {
const conf = new ConfigurationModule();
const chain_name = conf.config.settings.chain;
const env: WorkerEnv = {
worker_id: process.env.worker_id,
worker_role: process.env.worker_role
};
process.on('SIGINT', function () {
hLog("caught interrupt signal. Exiting now!");
process.exit();
});
if (cluster.isMaster) {
process.title = `hyp-${chain_name}-master`;
const master = await import('./modules/master');
new master.HyperionMaster().runMaster().catch((err) => {
console.log(process.env['worker_role']);
console.log(err);
});
} else {
if (hyperionWorkers[env.worker_role]) {
process.title = `hyp-${chain_name}-${env.worker_role}:${env.worker_id}`;
const mod = (await import(`./workers/${hyperionWorkers[env.worker_role]}`)).default;
const instance = new mod() as HyperionWorker;
await instance.run();
} else {
console.log(`FATAL: Unlisted Worker: ${env.worker_role}`);
}
}
}
+5
View File
@@ -20,6 +20,7 @@ export interface ScalingConfigs {
}
export interface MainSettings {
process_prefix?: string;
index_partition_size: number;
ignore_snapshot?: boolean;
ship_request_rev: string;
@@ -61,6 +62,7 @@ export interface IndexerConfigs {
process_deltas: boolean;
repair_mode: boolean;
max_inline: number;
disable_delta_rm?: boolean;
}
interface ApiLimits {
@@ -78,6 +80,9 @@ interface ApiLimits {
}
interface ApiConfigs {
disable_rate_limit?: boolean;
disable_tx_cache?: boolean;
tx_cache_expiration_sec?: number | string;
rate_limit_rpm?: number;
rate_limit_allow?: string[];
custom_core_token?: string;
+2
View File
@@ -1,9 +1,11 @@
interface AmqpConfig {
host: string;
api: string;
protocol: string;
user: string;
pass: string;
vhost: string;
frameMax: string;
}
interface ESConfig {
+50 -2
View File
@@ -1,5 +1,53 @@
import {launch} from "./hyperion-launcher";
import cluster from "cluster";
import {ConfigurationModule} from "./modules/config";
import {HyperionWorker} from "./workers/hyperionWorker";
import {hLog} from "./helpers/common_functions";
interface WorkerEnv {
worker_role: string;
worker_id: string;
}
const hyperionWorkers = {
ds_pool_worker: 'ds-pool',
reader: 'state-reader',
deserializer: 'deserializer',
continuous_reader: 'state-reader',
ingestor: 'indexer',
router: 'ws-router',
delta_updater: 'delta-updater'
};
async function launch() {
const conf = new ConfigurationModule();
const chain_name = conf.config.settings.chain;
const env: WorkerEnv = {
worker_id: process.env.worker_id,
worker_role: process.env.worker_role
};
process.on('SIGINT', function () {
hLog("caught interrupt signal. Exiting now!");
process.exit();
});
if (cluster.isPrimary) {
process.title = `${conf.proc_prefix}-${chain_name}-master`;
const master = await import('./modules/master');
new master.HyperionMaster().runMaster().catch((err) => {
console.log(process.env['worker_role']);
console.log(err);
});
} else {
if (hyperionWorkers[env.worker_role] && !conf.disabledWorkers.has(env.worker_role)) {
process.title = `${conf.proc_prefix}-${chain_name}-${env.worker_role}:${env.worker_id}`;
const mod = (await import(`./workers/${hyperionWorkers[env.worker_role]}`)).default;
const instance = new mod() as HyperionWorker;
await instance.run();
}
}
}
launch().catch((err) => {
console.error(err);
console.error(err);
});
+116 -104
View File
@@ -3,124 +3,136 @@ import {HyperionConnections} from "../interfaces/hyperionConnections";
import {HyperionConfig} from "../interfaces/hyperionConfig";
export interface Filters {
action_blacklist: Set<string>;
action_whitelist: Set<string>;
delta_whitelist: Set<string>;
delta_blacklist: Set<string>;
action_blacklist: Set<string>;
action_whitelist: Set<string>;
delta_whitelist: Set<string>;
delta_blacklist: Set<string>;
}
export class ConfigurationModule {
public config: HyperionConfig;
public connections: HyperionConnections;
EOSIO_ALIAS: string;
public config: HyperionConfig;
public connections: HyperionConnections;
public EOSIO_ALIAS: string;
public filters: Filters = {
action_blacklist: new Set(),
action_whitelist: new Set(),
delta_whitelist: new Set(),
delta_blacklist: new Set()
};
public filters: Filters = {
action_blacklist: new Set(),
action_whitelist: new Set(),
delta_whitelist: new Set(),
delta_blacklist: new Set()
};
constructor() {
this.loadConfigJson();
this.loadConnectionsJson();
}
public disabledWorkers = new Set();
public proc_prefix = 'hyp';
processConfig() {
// enforce deltas only for abi scan mode
if (this.config.indexer.abi_scan_mode) {
this.config.indexer.fetch_traces = false;
this.config.indexer.fetch_block = false;
}
constructor() {
this.loadConfigJson();
this.loadConnectionsJson();
}
this.EOSIO_ALIAS = 'eosio';
if (this.config.settings.eosio_alias) {
this.EOSIO_ALIAS = this.config.settings.eosio_alias;
} else {
this.config.settings.eosio_alias = 'eosio';
}
processConfig() {
// append default blacklists (eosio::onblock & eosio.null)
// this.filters.action_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}::onblock`);
this.filters.action_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}.null::*`);
if (this.config.settings.process_prefix) {
this.proc_prefix = this.config.settings.process_prefix;
}
// this.filters.delta_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}::global`);
// this.filters.delta_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}::global2`);
// this.filters.delta_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}::global3`);
if (this.config.indexer.disable_delta_rm) {
this.disabledWorkers.add('delta_updater');
}
// append user blacklists
if (this.config.blacklists) {
if (this.config.blacklists.actions) {
this.config.blacklists.actions.forEach((a) => {
this.filters.action_blacklist.add(a);
});
}
if (this.config.blacklists.deltas) {
this.config.blacklists.deltas.forEach((d) => {
this.filters.delta_blacklist.add(d);
});
}
}
// enforce deltas only for abi scan mode
if (this.config.indexer.abi_scan_mode) {
this.config.indexer.fetch_traces = false;
this.config.indexer.fetch_block = false;
}
// append user whitelists
if (this.config.whitelists) {
if (this.config.whitelists.actions) {
this.config.whitelists.actions.forEach((a) => {
this.filters.action_whitelist.add(a);
});
}
if (this.config.whitelists.deltas) {
this.config.whitelists.deltas.forEach((d) => {
this.filters.delta_whitelist.add(d);
});
}
this.EOSIO_ALIAS = 'eosio';
if (this.config.settings.eosio_alias) {
this.EOSIO_ALIAS = this.config.settings.eosio_alias;
} else {
this.config.settings.eosio_alias = 'eosio';
}
if (!this.config.whitelists.root_only) {
this.config.whitelists.root_only = false;
}
}
}
// append default blacklists (eosio::onblock & eosio.null)
// this.filters.action_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}::onblock`);
this.filters.action_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}.null::*`);
loadConfigJson() {
if (process.env.CONFIG_JSON) {
const data = readFileSync(process.env.CONFIG_JSON).toString();
try {
this.config = JSON.parse(data);
this.processConfig();
} catch (e) {
console.log(`Failed to Load configuration file ${process.env.CONFIG_JSON}`);
console.log(e);
process.exit(1);
}
} else {
console.error('Configuration file not specified!');
process.exit(1);
}
}
// this.filters.delta_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}::global`);
// this.filters.delta_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}::global2`);
// this.filters.delta_blacklist.add(`${this.config.settings.chain}::${this.EOSIO_ALIAS}::global3`);
setAbiScanMode(value: boolean) {
const data = readFileSync(process.env.CONFIG_JSON).toString();
const tempConfig: HyperionConfig = JSON.parse(data);
tempConfig.indexer.abi_scan_mode = value;
writeFileSync(process.env.CONFIG_JSON, JSON.stringify(tempConfig, null, 2));
this.config.indexer.abi_scan_mode = value;
}
// append user blacklists
if (this.config.blacklists) {
if (this.config.blacklists.actions) {
this.config.blacklists.actions.forEach((a) => {
this.filters.action_blacklist.add(a);
});
}
if (this.config.blacklists.deltas) {
this.config.blacklists.deltas.forEach((d) => {
this.filters.delta_blacklist.add(d);
});
}
}
loadConnectionsJson() {
const file = './connections.json';
if (existsSync(file)) {
const data = readFileSync(file).toString();
try {
this.connections = JSON.parse(data);
} catch (e) {
console.log(`Failed to Load ${file}`);
console.log(e.message);
process.exit(1);
}
} else {
console.log('connections.json not found!');
process.exit(1);
}
}
// append user whitelists
if (this.config.whitelists) {
if (this.config.whitelists.actions) {
this.config.whitelists.actions.forEach((a) => {
this.filters.action_whitelist.add(a);
});
}
if (this.config.whitelists.deltas) {
this.config.whitelists.deltas.forEach((d) => {
this.filters.delta_whitelist.add(d);
});
}
if (!this.config.whitelists.root_only) {
this.config.whitelists.root_only = false;
}
}
}
loadConfigJson() {
if (process.env.CONFIG_JSON) {
const data = readFileSync(process.env.CONFIG_JSON).toString();
try {
this.config = JSON.parse(data);
this.processConfig();
} catch (e) {
console.log(`Failed to Load configuration file ${process.env.CONFIG_JSON}`);
console.log(e);
process.exit(1);
}
} else {
console.error('Configuration file not specified!');
process.exit(1);
}
}
setAbiScanMode(value: boolean) {
const data = readFileSync(process.env.CONFIG_JSON).toString();
const tempConfig: HyperionConfig = JSON.parse(data);
tempConfig.indexer.abi_scan_mode = value;
writeFileSync(process.env.CONFIG_JSON, JSON.stringify(tempConfig, null, 2));
this.config.indexer.abi_scan_mode = value;
}
loadConnectionsJson() {
const file = './connections.json';
if (existsSync(file)) {
const data = readFileSync(file).toString();
try {
this.connections = JSON.parse(data);
} catch (e) {
console.log(`Failed to Load ${file}`);
console.log(e.message);
process.exit(1);
}
} else {
console.log('connections.json not found!');
process.exit(1);
}
}
}
+1 -1
View File
@@ -1,5 +1,5 @@
import DSPoolWorker from "../../workers/ds-pool";
import * as flatstr from 'flatstr';
import flatstr from 'flatstr';
export async function parseDSPEvent(worker: DSPoolWorker, data: any) {
const parsedEvents = [];
+166 -99
View File
@@ -16,7 +16,7 @@ import {
} from "../helpers/common_functions";
import {GetInfoResult} from "eosjs/dist/eosjs-rpc-interfaces";
import * as pm2io from '@pm2/io';
import pm2io from '@pm2/io';
import {
createWriteStream,
@@ -29,9 +29,8 @@ import {
WriteStream
} from "fs";
import * as path from "path";
import * as cluster from "cluster";
import {Worker} from "cluster";
import path from "path";
import cluster, {Worker} from "cluster";
import {io, Socket} from 'socket.io-client';
import {HyperionWorkerDef} from "../interfaces/hyperionWorkerDef";
import {HyperionConfig} from "../interfaces/hyperionConfig";
@@ -41,8 +40,9 @@ import {convertLegacyPublicKey} from "eosjs/dist/eosjs-numeric";
import moment = require("moment");
import Timeout = NodeJS.Timeout;
import AlertsManager from "./alertsManager";
import * as Redis from 'ioredis';
import IORedis from "ioredis";
import {IOConfig} from "@pm2/io/build/main/pmx";
import Gauge from "@pm2/io/build/main/utils/metrics/gauge";
interface RevBlock {
num: number;
@@ -87,7 +87,7 @@ export class HyperionMaster {
private dsErrorStream: WriteStream;
// mem-optimized deserialization pool
private dsPoolMap: Map<number, cluster.Worker> = new Map();
private dsPoolMap: Map<number, Worker> = new Map();
private globalUsageMap = {};
private totalContractHits = 0;
@@ -134,14 +134,14 @@ export class HyperionMaster {
// IPC Messages Handling
private msgHandlerMap: any;
private cachedInitABI = false;
private cachedInitABI = null;
private activeReadersCount = 0;
private lastAssignedBlock: number;
private lastIndexedABI: number;
private activeSchedule: any;
private pendingSchedule: any;
private proposedSchedule: any;
private wsRouterWorker: cluster.Worker;
private wsRouterWorker: Worker;
private liveBlockQueue: QueueObject<any>;
private readingPaused = false;
private readingLimited = false;
@@ -164,6 +164,10 @@ export class HyperionMaster {
private revBlockArray: RevBlock[] = [];
private shutdownStarted = false;
// pm2 custom metrics/gauges
private metrics: Record<string, Gauge> = {};
private shouldCountIPCMessages = false;
constructor() {
this.cm = new ConfigurationModule();
this.conf = this.cm.config;
@@ -361,7 +365,7 @@ export class HyperionMaster {
b: msg.block_num,
s: msg.signatures,
a: msg.root_act
}), "EX", 600).catch(console.log);
}), "EX", 3600).catch(console.log);
}
},
'fork_event': async (msg: any) => {
@@ -1005,24 +1009,12 @@ export class HyperionMaster {
}
handleMessage(msg) {
if (this.conf.settings.ipc_debug_rate && this.conf.settings.ipc_debug_rate >= 1000) {
if (this.shouldCountIPCMessages) {
this.totalMessages++;
}
if (this.msgHandlerMap[msg.event]) {
this.msgHandlerMap[msg.event](msg);
} else {
if (msg.type) {
if (msg.type === 'axm:monitor') {
if (process.env['AXM_DEBUG'] === 'true') {
hLog(`----------- axm:monitor ------------`);
for (const key in msg.data) {
if (msg.data.hasOwnProperty(key)) {
hLog(`${key}: ${msg.data[key].value}`);
}
}
}
}
}
}
}
@@ -1089,7 +1081,7 @@ export class HyperionMaster {
}
};
this.client.index({
index: this.chain + '-logs',
index: this.chain + '-logs-' + this.conf.settings.index_version,
body: _body
}).catch(hLog);
}
@@ -1288,6 +1280,7 @@ export class HyperionMaster {
const testedQueues = new Set();
let aboveLimit = false;
let canResume = true;
let queuedMessages = 0;
for (const worker of this.workerMap) {
let queue = worker.queue;
if (!queue) {
@@ -1301,6 +1294,7 @@ export class HyperionMaster {
if (queue && !testedQueues.has(queue)) {
const size = await this.manager.checkQueueSize(queue);
queuedMessages += size;
let qlimit = this.conf.scaling.max_queue_limit;
if (worker.worker_role === 'deserializer') {
qlimit = this.conf.scaling.block_queue_limit;
@@ -1384,6 +1378,7 @@ export class HyperionMaster {
this.resumeReaders();
}
}
this.metrics.queuedMessages?.set(queuedMessages);
}
private monitorIndexingQueues() {
@@ -1402,34 +1397,6 @@ export class HyperionMaster {
}
}
private onPm2Stop() {
pm2io.action('stop', (reply) => {
try {
this.shutdownStarted = true;
this.allowMoreReaders = false;
const stopMsg = 'Stop signal received. Shutting down readers immediately!';
hLog(stopMsg);
this.emitAlert('warning', stopMsg);
messageAllWorkers(cluster, {
event: 'stop'
});
setInterval(() => {
if (this.allowShutdown) {
getLastIndexedBlockFromRange(this.client, this.chain, this.starting_block, this.head).then((value: number) => {
hLog(`Last Indexed Block: ${value}`);
writeFileSync(`./chains/.${this.chain}_lastblock.txt`, value.toString());
hLog('Shutting down master...');
reply({ack: true});
process.exit(1);
});
}
}, 100);
} catch (e) {
reply({ack: false, error: e.message});
}
});
}
private startIndexMonitoring() {
let reference_time = Date.now();
@@ -1463,15 +1430,23 @@ export class HyperionMaster {
const _r = (this.pushedBlocks + this.livePushedBlocks) / tScale;
log_msg.push(`R:${_r}`);
this.metrics.readingRate?.set(_r);
const _c = (this.liveConsumedBlocks + this.consumedBlocks) / tScale;
log_msg.push(`C:${_c}`);
this.metrics.consumeRate?.set(_c);
const _a = (this.deserializedActions) / tScale;
log_msg.push(`A:${_a}`);
this.metrics.actionDSRate?.set(_a);
log_msg.push(`D:${(this.deserializedDeltas) / tScale}`);
log_msg.push(`I:${this.indexedObjects / tScale}`);
const _dds = (this.deserializedDeltas) / tScale
log_msg.push(`D:${_dds}`);
this.metrics.deltaDSRate?.set(_dds);
const _ir = (this.indexedObjects) / tScale
log_msg.push(`I:${_ir}`);
this.metrics.indexingRate?.set(_ir);
if (this.total_blocks < this.total_range && !this.conf.indexer.live_only_mode) {
const remaining = this.total_range - this.total_blocks;
@@ -1481,8 +1456,12 @@ export class HyperionMaster {
const pct_read = ((this.total_read / this.total_range) * 100).toFixed(1);
log_msg.push(`${this.total_blocks}/${this.total_read}/${this.total_range}`);
log_msg.push(`syncs ${time_string} (${pct_parsed}% ${pct_read}%)`);
this.metrics.syncEta.set(time_string);
}
// publish last processed block to pm2
this.metrics.lastProcessedBlockNum.set(this.lastProcessedBlockNum);
// publish log to hub
if (this.conf.hub && this.conf.hub.inform_url) {
this.hub.emit('hyp_ev', {
@@ -1708,7 +1687,7 @@ export class HyperionMaster {
}
// Redis
this.ioRedisClient = new Redis(this.manager.conn.redis);
this.ioRedisClient = new IORedis(this.manager.conn.redis);
// Elasticsearch
this.client = this.manager.elasticsearchClient;
@@ -1793,6 +1772,8 @@ export class HyperionMaster {
await this.setupWorkers();
// TODO: fix repair mode
// Run and wait for repair
if (this.conf.indexer.repair_mode) {
await this.startRepairMode();
}
@@ -1809,6 +1790,7 @@ export class HyperionMaster {
// Start Monitoring
this.startIndexMonitoring();
// handle worker disconnection events
cluster.on('disconnect', (worker) => {
if (!this.mode_transition && !this.shutdownStarted) {
hLog(`The worker #${worker.id} has disconnected, attempting to re-launch in 5 seconds...`);
@@ -1837,24 +1819,25 @@ export class HyperionMaster {
// Launch all workers
this.launchWorkers();
if (this.conf.settings.ipc_debug_rate > 0 && this.conf.settings.ipc_debug_rate < 1000) {
// enable ipc msg rate monitoring
let ipcDebugRate = this.conf.settings.ipc_debug_rate;
if (ipcDebugRate > 0 && ipcDebugRate < 1000) {
hLog(`settings.ipc_debug_rate was set too low (${this.conf.settings.ipc_debug_rate}) using 1000 instead!`);
this.conf.settings.ipc_debug_rate = 1000;
ipcDebugRate = 1000;
}
this.shouldCountIPCMessages = ipcDebugRate && ipcDebugRate >= 1000;
if (this.conf.settings.debug) {
if (this.conf.settings.ipc_debug_rate && this.conf.settings.ipc_debug_rate >= 1000) {
const rate = this.conf.settings.ipc_debug_rate;
if (this.shouldCountIPCMessages) {
const rate = ipcDebugRate;
this.totalMessages = 0;
setInterval(() => {
hLog(`IPC Messaging Rate: ${(this.totalMessages / (rate / 1000)).toFixed(2)} msg/s`);
this.totalMessages = 0;
}, rate);
}, ipcDebugRate);
}
}
this.onPm2Stop();
// load hyperion hub connection
if (this.conf.hub) {
try {
this.startHyperionHub();
@@ -1863,43 +1846,11 @@ export class HyperionMaster {
}
}
pm2io.action('get_usage_map', (reply) => {
reply(this.globalUsageMap);
});
// Actions that can be triggered via pm2 trigger <action> <process> or on the pm2 dashboard
this.addPm2Actions();
pm2io.action('get_heap', (reply) => {
const requests = [];
for (const id in cluster.workers) {
if (cluster.workers.hasOwnProperty(id)) {
const worker: Worker = cluster.workers[id];
requests.push(new Promise((resolve) => {
const _timeout = setTimeout(() => {
worker.removeListener('message', _listener);
resolve([id, null]);
}, 1000);
const _listener = (msg) => {
if (msg.event === 'v8_heap_report') {
clearTimeout(_timeout);
worker.removeListener('message', _listener);
resolve([msg.id, msg.data]);
}
};
worker.on('message', _listener);
worker.send({event: 'request_v8_heap_stats'});
}));
}
}
Promise.all(requests).then((results) => {
const responses = {};
for (const result of results) {
if (result[1]) {
responses[result[0]] = result[1];
}
}
reply(responses);
});
});
// Add custom metrics
this.addPm2Metrics();
}
async streamBlockHeaders(start_on, stop_on, func: (data) => void) {
@@ -2282,5 +2233,121 @@ export class HyperionMaster {
private emitAlert(msgType: string, msgContent: string) {
this.alerts.emitAlert({type: msgType, process: process.title, content: msgContent});
}
private addPm2Metrics() {
this.metrics.readingRate = pm2io.metric({name: 'Block Reading (blocks/s)'});
this.metrics.consumeRate = pm2io.metric({name: 'Block Processing (blocks/s)'});
this.metrics.actionDSRate = pm2io.metric({name: 'Action Deserialization (actions/s)'});
this.metrics.deltaDSRate = pm2io.metric({name: 'Delta Deserialization (deltas/s)'});
this.metrics.indexingRate = pm2io.metric({name: 'Indexing Rate (docs/s)'});
this.metrics.queuedMessages = pm2io.metric({name: 'Queued Messages'});
this.metrics.lastProcessedBlockNum = pm2io.metric({name: 'Last Block'});
this.metrics.syncEta = pm2io.metric({name: 'Time to sync'});
}
private addPm2Actions() {
const ioConfig: IOConfig = {
apmOptions: {
appName: this.chain + " indexer",
publicKey: undefined,
secretKey: undefined,
sendLogs: true
},
metrics: {
v8: true,
http: false,
network: false,
runtime: true,
eventLoop: true
}
};
pm2io.init(ioConfig);
pm2io.action('stop', (reply) => {
try {
this.shutdownStarted = true;
this.allowMoreReaders = false;
const stopMsg = 'Stop signal received. Shutting down readers immediately!';
hLog(stopMsg);
this.emitAlert('warning', stopMsg);
messageAllWorkers(cluster, {
event: 'stop'
});
setInterval(() => {
if (this.allowShutdown) {
getLastIndexedBlockFromRange(this.client, this.chain, this.starting_block, this.head).then((value: number) => {
hLog(`Last Indexed Block: ${value}`);
writeFileSync(`./chains/.${this.chain}_lastblock.txt`, value.toString());
hLog('Shutting down master...');
reply({ack: true});
process.exit(1);
});
}
}, 100);
} catch (e) {
reply({ack: false, error: e.message});
}
});
pm2io.action('get_usage_map', (reply) => {
reply(this.globalUsageMap);
});
pm2io.action('get_memory_usage', (reply) => {
this.requestDataFromWorkers(
{
event: 'request_memory_usage'
},
'memory_report'
).then(value => {
reply(value);
});
});
pm2io.action('get_heap', (reply) => {
this.requestDataFromWorkers(
{
event: 'request_v8_heap_stats'
},
'v8_heap_report'
).then(value => {
reply(value);
});
});
}
async requestDataFromWorkers(requestEvent: { event: string, data?: any }, responseEventType: string, timeoutSec: number = 1000) {
const requests = [];
for (const id in cluster.workers) {
if (cluster.workers.hasOwnProperty(id)) {
const worker: Worker = cluster.workers[id];
requests.push(new Promise((resolve) => {
const _timeout = setTimeout(() => {
worker.removeListener('message', _listener);
resolve([id, null]);
}, timeoutSec);
const _listener = (msg) => {
if (msg.event === responseEventType) {
clearTimeout(_timeout);
worker.removeListener('message', _listener);
resolve([msg.id, msg.data]);
}
};
worker.on('message', _listener);
worker.send(requestEvent);
}));
}
}
const results = await Promise.all(requests);
const responses = {};
for (const result of results) {
if (result[1]) {
responses[result[0]] = result[1];
}
}
return responses;
}
}
+231 -196
View File
@@ -8,244 +8,279 @@ import {deserialize, hLog} from "../../helpers/common_functions";
import {appendFileSync} from "fs";
function timedFunction(enable: boolean, method: () => void) {
if (enable) {
const ref = process.hrtime.bigint();
method();
return Number(process.hrtime.bigint() - ref) / 1000;
} else {
method();
return null;
}
if (enable) {
const ref = process.hrtime.bigint();
method();
return Number(process.hrtime.bigint() - ref) / 1000;
} else {
method();
return null;
}
}
export default class HyperionParser extends BaseParser {
flatten = false;
flatten = false;
public async parseAction(worker: DSPoolWorker, ts, action: ActionTrace, trx_data: TrxMetadata, _actDataArray, _processedTraces: ActionTrace[], full_trace, usageIncluded): Promise<boolean> {
// check filters
if (this.checkBlacklist(action.act)) return false;
if (this.filters.action_whitelist.size > 0) {
if (!this.checkWhitelist(action.act)) return false;
}
public async parseAction(worker: DSPoolWorker, ts, action: ActionTrace, trx_data: TrxMetadata, _actDataArray, _processedTraces: ActionTrace[], full_trace, usageIncluded): Promise<boolean> {
// check filters
if (this.checkBlacklist(action.act)) return false;
if (this.filters.action_whitelist.size > 0) {
if (!this.checkWhitelist(action.act)) return false;
}
await this.deserializeActionData(worker, action, trx_data);
await this.deserializeActionData(worker, action, trx_data);
action["@timestamp"] = ts;
action.block_num = trx_data.block_num;
action.block_id = trx_data.block_id;
action.producer = trx_data.producer;
action.trx_id = trx_data.trx_id;
action["@timestamp"] = ts;
action.block_num = trx_data.block_num;
action.block_id = trx_data.block_id;
action.producer = trx_data.producer;
action.trx_id = trx_data.trx_id;
if (action.account_ram_deltas.length === 0) {
delete action.account_ram_deltas;
}
if (action.account_ram_deltas.length === 0) {
delete action.account_ram_deltas;
}
if (action.except === null) {
if (!action.receipt) {
console.log(full_trace.status);
console.log(action);
}
if (action.except === null) {
if (!action.receipt) {
console.log(full_trace.status);
console.log(action);
}
action.receipt = action.receipt[1];
action.global_sequence = parseInt(action.receipt.global_sequence, 10);
action.receipt = action.receipt[1];
action.global_sequence = parseInt(action.receipt.global_sequence, 10);
delete action.except;
delete action.error_code;
delete action.except;
delete action.error_code;
// add usage data to the first action on the transaction
if (!usageIncluded.status) {
this.extendFirstAction(worker, action, trx_data, full_trace, usageIncluded);
}
// add usage data to the first action on the transaction
if (!usageIncluded.status) {
this.extendFirstAction(worker, action, trx_data, full_trace, usageIncluded);
}
_processedTraces.push(action);
} else {
hLog(action);
}
_processedTraces.push(action);
} else {
hLog(action);
}
return true;
}
return true;
}
public async parseMessage(worker: MainDSWorker, messages: Message[]): Promise<void> {
const dsProfiling = worker.conf.settings.ds_profiling;
for (const message of messages) {
public async parseMessage(worker: MainDSWorker, messages: Message[]): Promise<void> {
const dsProfiling = worker.conf.settings.ds_profiling;
for (const message of messages) {
// profile deserialization
const ds_times = {
size: message.content.length,
eosjs: {
result: undefined,
signed_block: undefined,
transaction_trace: undefined,
table_delta: undefined
},
abieos: {
result: undefined,
signed_block: undefined,
transaction_trace: undefined,
table_delta: undefined
}
};
// profile deserialization
const ds_times = {
size: message.content.length,
eosjs: {
result: undefined,
signed_block: undefined,
transaction_trace: undefined,
table_delta: undefined
},
abieos: {
result: undefined,
signed_block: undefined,
transaction_trace: undefined,
table_delta: undefined
}
};
let allowProcessing = true;
let ds_msg;
let allowProcessing = true;
let ds_msg;
// deserialize result using abieos
// ds_times.abieos.result = timedFunction(dsProfiling,() => {
// ds_msg = worker.deserializeNative('result', message.content);
// });
// deserialize result using abieos
// ds_times.abieos.result = timedFunction(dsProfiling,() => {
// ds_msg = worker.deserializeNative('result', message.content);
// });
// deserialize result using eosjs (faster)
ds_times.eosjs.result = timedFunction(dsProfiling, () => {
ds_msg = deserialize('result', message.content, this.txEnc, this.txDec, worker.types);
});
// deserialize result using eosjs (faster)
ds_times.eosjs.result = timedFunction(dsProfiling, () => {
ds_msg = deserialize('result', message.content, this.txEnc, this.txDec, worker.types);
});
if (!ds_msg) {
if (worker.ch_ready) {
worker.ch.nack(message);
throw new Error('failed to deserialize datatype=result');
}
}
if (!ds_msg) {
if (worker.ch_ready) {
worker.ch.nack(message);
throw new Error('failed to deserialize datatype=result');
}
}
const res = ds_msg[1];
let block = null;
let traces = [];
let deltas = [];
const res = ds_msg[1];
if (res.block && res.block.length) {
let block = null;
if (res.block && res.block.length) {
// deserialize signed_block using abieos (faster)
ds_times.abieos.signed_block = timedFunction(dsProfiling, () => {
block = worker.deserializeNative('signed_block_variant', res.block)[1];
});
if (typeof res.block === 'object' && res.block.length === 2) {
// already deserialized
block = res.block[1];
} else {
try {
// deserialize signed_block using abieos (faster)
ds_times.abieos.signed_block = timedFunction(dsProfiling, () => {
block = worker.deserializeNative('signed_block_variant', res.block)[1];
});
} catch (e) {
hLog('signed_block_variant deserialization failed with abieos!');
}
if (!block) {
try {
// deserialize signed_block using eosjs
ds_times.eosjs.signed_block = timedFunction(dsProfiling, () => {
block = deserialize('signed_block_variant', res.block, this.txEnc, this.txDec, worker.types);
});
} catch (e) {
hLog('signed_block_variant deserialization failed with eosjs!');
}
}
}
// // deserialize signed_block using eosjs
// ds_times.eosjs.signed_block = timedFunction(dsProfiling, () => {
// console.log(worker.types);
// block = deserialize('signed_block_variant', res.block, this.txEnc, this.txDec, worker.types);
// console.log(block);
// });
if (block === null) {
console.log(res);
process.exit(1);
}
if (block === null) {
console.log(res);
process.exit(1);
}
// verify for whitelisted contracts (root actions only)
if (worker.conf.whitelists.root_only) {
try {
// verify for whitelisted contracts (root actions only)
if (worker.conf.whitelists.root_only) {
try {
// get time reference for profiling
if (worker.conf.settings.ds_profiling) ds_times['packed_trx'] = process.hrtime.bigint();
// get time reference for profiling
if (worker.conf.settings.ds_profiling) ds_times['packed_trx'] = process.hrtime.bigint();
if (worker.conf.whitelists &&
(worker.conf.whitelists.actions.length > 0 ||
worker.conf.whitelists.deltas.length > 0)) {
if (worker.conf.whitelists &&
(worker.conf.whitelists.actions.length > 0 ||
worker.conf.whitelists.deltas.length > 0)) {
allowProcessing = false;
allowProcessing = false;
for (const transaction of block.transactions) {
if (transaction.status === 0 && transaction.trx[1] && transaction.trx[1].packed_trx) {
const unpacked_trx = worker.api.deserializeTransaction(Buffer.from(transaction.trx[1].packed_trx, 'hex'));
for (const act of unpacked_trx.actions) {
if (this.checkWhitelist(act)) {
allowProcessing = true;
break;
}
}
if (allowProcessing) break;
}
}
}
for (const transaction of block.transactions) {
if (transaction.status === 0 && transaction.trx[1] && transaction.trx[1].packed_trx) {
const unpacked_trx = worker.api.deserializeTransaction(Buffer.from(transaction.trx[1].packed_trx, 'hex'));
for (const act of unpacked_trx.actions) {
if (this.checkWhitelist(act)) {
allowProcessing = true;
break;
}
}
if (allowProcessing) break;
}
}
}
// store time diff
if (worker.conf.settings.ds_profiling) ds_times['packed_trx'] = Number(process.hrtime.bigint() - ds_times['packed_trx']) / 1000;
// store time diff
if (worker.conf.settings.ds_profiling) ds_times['packed_trx'] = Number(process.hrtime.bigint() - ds_times['packed_trx']) / 1000;
} catch (e) {
console.log(e);
allowProcessing = true;
}
}
}
} catch (e) {
console.log(e);
allowProcessing = true;
}
}
}
// unpack traces
let traces = null;
if (allowProcessing && res.traces && res.traces.length) {
if (allowProcessing && res.traces && res.traces.length) {
// deserialize transaction_trace using abieos (faster)
try {
ds_times.abieos.transaction_trace = timedFunction(dsProfiling, () => {
traces = worker.deserializeNative('transaction_trace[]', res.traces);
});
} catch (e) {
hLog('transaction_trace[] deserialization failed with abieos!');
}
// deserialize transaction_trace using abieos (faster)
ds_times.abieos.transaction_trace = timedFunction(dsProfiling, () => {
traces = worker.deserializeNative('transaction_trace[]', res.traces);
});
// deserialize transaction_trace using eosjs
if (!traces) {
try {
ds_times.eosjs.transaction_trace = timedFunction(dsProfiling, () => {
traces = deserialize('transaction_trace[]', res.traces, this.txEnc, this.txDec, worker.types);
});
} catch (e) {
hLog('transaction_trace[] deserialization failed with eosjs!');
}
}
// deserialize transaction_trace using eosjs
// ds_times.eosjs.transaction_trace = timedFunction(dsProfiling, () => {
// traces = deserialize('transaction_trace[]', res.traces, this.txEnc, this.txDec, worker.types);
// });
if (!traces) {
hLog(`[WARNING] transaction_trace[] deserialization failed on block ${res['this_block']['block_num']}`);
}
}
if (!traces) {
hLog(`[WARNING] transaction_trace[] deserialization failed on block ${res['this_block']['block_num']}`);
}
}
if (allowProcessing && res.deltas && res.deltas.length) {
// unpack deltas
let deltas = null;
if (allowProcessing && res.deltas && res.deltas.length) {
// deserialize table_delta using abieos
// ds_times.abieos.table_delta = timedFunction(dsProfiling, () => {
// worker.deserializeNative('table_delta[]', res.deltas);
// });
// deserialize table_delta using abieos
try {
ds_times.abieos.table_delta = timedFunction(dsProfiling, () => {
deltas = worker.deserializeNative('table_delta[]', res.deltas);
});
} catch (e) {
hLog('table_delta[] deserialization failed with abieos!');
}
// deserialize table_delta using eosjs (faster)
ds_times.eosjs.table_delta = timedFunction(dsProfiling, () => {
deltas = deserialize('table_delta[]', res.deltas, this.txEnc, this.txDec, worker.types);
});
// deserialize table_delta using eosjs
if (!deltas) {
try {
ds_times.eosjs.table_delta = timedFunction(dsProfiling, () => {
deltas = deserialize('table_delta[]', res.deltas, this.txEnc, this.txDec, worker.types);
});
} catch (e) {
hLog('table_delta[] deserialization failed with eosjs!');
}
}
if (!deltas) {
hLog(`[WARNING] table_delta[] deserialization failed on block ${res['this_block']['block_num']}`);
}
}
if (!deltas) {
hLog(`[WARNING] table_delta[] deserialization failed on block ${res['this_block']['block_num']}`);
}
}
if (worker.conf.settings.ds_profiling) {
hLog(ds_times);
const line = [ds_times.size];
line.push(...[ds_times.eosjs.result, ds_times.eosjs.signed_block, ds_times.eosjs.transaction_trace, ds_times.eosjs.table_delta]);
line.push(...[ds_times.abieos.result, ds_times.abieos.signed_block, ds_times.abieos.transaction_trace, ds_times.abieos.table_delta]);
appendFileSync('ds_profiling.csv', line.join(',') + "\n");
}
if (worker.conf.settings.ds_profiling) {
hLog(ds_times);
const line = [ds_times.size];
line.push(...[ds_times.eosjs.result, ds_times.eosjs.signed_block, ds_times.eosjs.transaction_trace, ds_times.eosjs.table_delta]);
line.push(...[ds_times.abieos.result, ds_times.abieos.signed_block, ds_times.abieos.transaction_trace, ds_times.abieos.table_delta]);
appendFileSync('ds_profiling.csv', line.join(',') + "\n");
}
let result;
try {
result = await worker.processBlock(res, block, traces, deltas);
if (result) {
const evPayload = {
event: 'consumed_block',
block_num: result['block_num'],
block_id: result['block_id'],
trx_ids: result['trx_ids'],
lib: res.last_irreversible.block_num,
live: process.env.live_mode
};
if (block) {
evPayload["producer"] = block['producer'];
evPayload["schedule_version"] = block['schedule_version'];
}
process.send(evPayload);
} else {
hLog(`ERROR: Block data not found for #${res['this_block']['block_num']}`);
}
if (worker.ch_ready) {
worker.ch.ack(message);
}
} catch (e) {
console.log(e);
if (worker.ch_ready) {
worker.ch.nack(message);
}
process.exit(1);
}
}
}
let result;
try {
result = await worker.processBlock(res, block, traces, deltas);
if (result) {
const evPayload = {
event: 'consumed_block',
block_num: result['block_num'],
block_id: result['block_id'],
trx_ids: result['trx_ids'],
lib: res.last_irreversible.block_num,
live: process.env.live_mode
};
if (block) {
evPayload["producer"] = block['producer'];
evPayload["schedule_version"] = block['schedule_version'];
}
process.send(evPayload);
} else {
hLog(`ERROR: Block data not found for #${res['this_block']['block_num']}`);
}
if (worker.ch_ready) {
worker.ch.ack(message);
}
} catch (e) {
console.log(e);
if (worker.ch_ready) {
worker.ch.nack(message);
}
process.exit(1);
}
}
}
async flattenInlineActions(action_traces: any[]): Promise<any[]> {
hLog(`Calling undefined flatten operation!`);
return Promise.resolve(undefined);
}
async flattenInlineActions(action_traces: any[]): Promise<any[]> {
hLog(`Calling undefined flatten operation!`);
return Promise.resolve(undefined);
}
}
+491 -477
View File
File diff suppressed because it is too large Load Diff
+34 -31
View File
@@ -1,6 +1,6 @@
{
"name": "hyperion-history",
"version": "3.3.4-rc4",
"version": "3.3.4-rc7",
"description": "Scalable Full History API Solution for EOSIO based blockchains",
"main": "launcher.js",
"scripts": {
@@ -10,9 +10,12 @@
"tsc": "tsc",
"build": "tsc",
"postinstall": "npm run build && npm run fix-permissions",
"fix-permissions": "chmod u+x run.sh stop.sh install_env.sh",
"fix-permissions": "node scripts/fix-permissions.js",
"plugin-manager": "node plugins/hyperion-plugin-manager.js"
},
"engines": {
"node": "^16"
},
"author": {
"name": "EOS Rio",
"url": "https://eosrio.io"
@@ -23,49 +26,49 @@
},
"license": "MIT",
"dependencies": {
"@elastic/elasticsearch": "^7.13.0",
"@elastic/elasticsearch": "7.14.1",
"@eosrio/node-abieos": "^2.1.1",
"@pm2/io": "^5.0.0",
"amqplib": "^0.8.0",
"async": "3.2.0",
"async": "^3.2.1",
"base-x": "^3.0.8",
"cross-fetch": "^3.1.4",
"eosjs": "^22.0.0",
"fast-json-stringify": "^2.7.6",
"fastify": "^3.17.0",
"fastify-autoload": "^3.7.1",
"fastify-compress": "^3.5.0",
"fastify-cors": "^6.0.1",
"fastify-elasticsearch": "2.0.0",
"fastify-formbody": "^5.0.0",
"eosjs": "^22.1.0",
"fast-json-stringify": "^2.7.9",
"fastify": "^3.20.2",
"fastify-autoload": "^3.8.1",
"fastify-compress": "^3.6.0",
"fastify-cors": "^6.0.2",
"fastify-elasticsearch": "^2.0.0",
"fastify-formbody": "^5.1.0",
"fastify-plugin": "^3.0.0",
"fastify-rate-limit": "^5.5.0",
"fastify-redis": "^4.3.0",
"fastify-static": "^4.2.2",
"fastify-swagger": "^4.7.0",
"fastify-rate-limit": "^5.6.2",
"fastify-redis": "^4.3.1",
"fastify-static": "^4.2.3",
"fastify-swagger": "^4.9.1",
"flatstr": "^1.0.12",
"got": "11.8.2",
"ioredis": "^4.27.6",
"ioredis": "^4.27.9",
"lodash": "^4.17.21",
"moment": "^2.29.1",
"nodemailer": "^6.6.1",
"nodemailer": "^6.6.3",
"portfinder": "^1.0.28",
"socket.io": "^4.1.2",
"socket.io-client": "^4.1.2",
"socket.io": "^4.2.0",
"socket.io-client": "^4.2.0",
"socket.io-redis": "^6.1.1",
"telegraf": "^4.3.0",
"typescript": "^4.3.2",
"ws": "^7.4.6",
"yargs": "^17.0.1"
"telegraf": "^4.4.1",
"typescript": "^4.3.5",
"ws": "^8.2.1",
"yargs": "^17.1.1"
},
"devDependencies": {
"@types/amqplib": "^0.8.0",
"@types/async": "^3.2.6",
"@types/ioredis": "^4.26.4",
"@types/lodash": "^4.14.170",
"@types/node": "^15.12.2",
"@types/nodemailer": "^6.4.2",
"@types/ws": "^7.4.4"
"@types/amqplib": "^0.8.2",
"@types/async": "^3.2.7",
"@types/ioredis": "^4.27.2",
"@types/lodash": "^4.14.172",
"@types/node": "^16.7.10",
"@types/nodemailer": "^6.4.4",
"@types/ws": "^7.4.7"
},
"optionalDependencies": {
"bufferutil": "^4.0.3",
+12 -3
View File
@@ -4,8 +4,17 @@ if [ $# -eq 0 ]; then
exit 1
fi
echo -e "\n-->> Starting $1..."
(set -x; pm2 start --only "$@" --update-env --silent)
(
set -x
pm2 start --only "$@" --update-env --silent
)
echo -e "\n-->> Saving pm2 state..."
(set -x; pm2 save)
(
set -x
pm2 save
)
echo -e "\n-->> Reading $1 logs..."
(set -x; pm2 logs --raw --lines 10 "$@")
(
set -x
pm2 logs --raw --lines 10 "$@"
)
+9
View File
@@ -0,0 +1,9 @@
const os = require('os');
const {execSync} = require('child_process');
if (os.type() === 'Windows_NT') {
console.log('Skipping permission update on windows');
} else {
console.log('Updating permissions...');
execSync('chmod u+x run.sh stop.sh');
}
+2 -1
View File
@@ -1,7 +1,8 @@
{
"compilerOptions": {
"module": "commonjs",
"target": "ES2019",
"target": "ES2020",
"esModuleInterop": true,
"allowSyntheticDefaultImports": true,
"noImplicitAny": false,
"moduleResolution": "node",
+1
View File
@@ -16,5 +16,6 @@ declare module 'fastify' {
push_api: string;
tokenCache: Map<string, any>;
allowedActionQueryParamSet: Set<string>;
routeSet: Set<string>;
}
}
+115 -79
View File
@@ -1,94 +1,130 @@
import {HyperionWorker} from "./hyperionWorker";
import {hLog} from "../helpers/common_functions";
import {Message} from "amqplib";
import {createHash} from "crypto";
interface HyperionDelta {
"@timestamp": string;
code: string;
scope: string;
table: string;
primary_key: string;
block_num: number;
block_id: string;
data: any;
"@timestamp": string;
code: string;
scope: string;
table: string;
primary_key: string;
block_num: number;
block_id: string;
data: any;
}
// noinspection JSUnusedGlobalSymbols
export default class MainDSWorker extends HyperionWorker {
constructor() {
super();
}
assertQueues(): void {
if (this.ch) {
const queue = this.chain + ":delta_rm";
this.ch.assertQueue(queue, {durable: true});
this.ch.prefetch(1);
this.ch.consume(queue, this.onConsume.bind(this));
}
}
queueName!: string;
retryMap = new Map<string, number>();
retryUpdates = false;
onConsume(data: Message) {
const deltaData = JSON.parse(Buffer.from(data.content).toString());
this.updateDelta(deltaData).then((status) => {
if (status) {
this.ch.ack(data);
} else {
this.ch.nack(data, false, true);
}
}).catch((err) => {
console.log(err);
this.ch.nack(data);
});
}
constructor() {
super();
}
async updateDelta(delta: HyperionDelta): Promise<boolean> {
try {
const searchResult = await this.manager.elasticsearchClient.updateByQuery({
index: this.chain + '-delta-*',
conflicts: "proceed",
body: {
query: {
bool: {
must: [
{term: {"code": {"value": delta.code}}},
{term: {"table": {"value": delta.table}}},
{term: {"scope": {"value": delta.scope}}},
{term: {"primary_key": {"value": delta.primary_key}}},
{range: {"block_num": {"lte": delta.block_num}}}
],
must_not: [{exists: {field: "deleted_at"}}]
}
},
script: {
source: "ctx._source.deleted_at = params.blk",
lang: "painless",
params: {
"blk": delta.block_num
}
}
}
});
if (searchResult.body.total === 0) {
// no results, send back to queue
// hLog(`Updated Deltas: ${searchResult.body.updated}`);
// hLog(delta);
await new Promise(resolve => setTimeout(resolve, 100));
return false;
} else {
hLog(`Updated Deltas: ${searchResult.body.updated}`);
return true;
}
} catch (e) {
hLog(e.message);
return false;
}
}
assertQueues(): void {
if (this.ch) {
this.queueName = this.chain + ":delta_rm";
hLog(`Launched delta updater, consuming from ${this.queueName}`);
this.ch.assertQueue(this.queueName, {durable: true});
this.ch.prefetch(1);
this.ch.consume(this.queueName, this.onConsume.bind(this));
}
}
onIpcMessage(msg: any): void {
}
onConsume(data: Message) {
const deltaData = JSON.parse(Buffer.from(data.content).toString());
this.updateDelta(deltaData).then((status) => {
if (status) {
this.ch.ack(data);
} else {
this.ch.nack(data, false, true);
}
}).catch((err) => {
console.log(err);
this.ch.nack(data);
});
}
async run(): Promise<void> {
hLog('Launching delta updater!');
}
getIndexPartition(blockNum: number): string {
return Math.ceil(blockNum / this.conf.settings.index_partition_size).toString().padStart(6, '0');
}
async updateDelta(delta: HyperionDelta): Promise<boolean> {
// TODO: update only on specific blockrange
// console.log(delta.block_num, '-->', this.getIndexPartition(delta.block_num));
const identifier = `${delta.code}-${delta.table}-${delta.scope}-${delta.primary_key}-${delta.block_num}`;
try {
const updateByQuery = {
query: {
bool: {
must: [
{term: {"code": {"value": delta.code}}},
{term: {"table": {"value": delta.table}}},
{term: {"scope": {"value": delta.scope}}},
{term: {"primary_key": {"value": delta.primary_key}}},
{range: {"block_num": {"lte": delta.block_num}}}
],
must_not: [{exists: {field: "deleted_at"}}]
}
},
script: {
source: "ctx._source.deleted_at = params.blk",
lang: "painless",
params: {
"blk": delta.block_num
}
}
};
const searchResult = await this.manager.elasticsearchClient.updateByQuery({
index: this.chain + '-delta-*',
conflicts: "proceed",
body: updateByQuery
});
if (searchResult.body.total === 0) {
if (!this.retryUpdates) {
// hLog(`Skipping ${identifier}`);
return true;
}
const hash = createHash('sha256')
.update(identifier)
.digest('hex');
if (!this.retryMap.has(hash)) {
this.retryMap.set(hash, 1);
} else {
const retries = this.retryMap.get(hash);
if (retries > 3) {
// abort update after 3 attempts
hLog(`Hyperion wasn't able to update previous deltas for ${identifier}`);
this.retryMap.delete(hash);
return true;
} else {
this.retryMap.set(hash, retries + 1);
}
}
await new Promise(resolve => setTimeout(resolve, 100));
return false;
} else {
if (searchResult.body.took > 500) {
hLog(`Updated Deltas: ${searchResult.body.updated} | Took: ${searchResult.body.took} ms`);
hLog(identifier);
}
return true;
}
} catch (e) {
hLog(e.message);
return false;
}
}
onIpcMessage(msg: any): void {
}
async run(): Promise<void> {
}
}
+1719 -1705
View File
File diff suppressed because it is too large Load Diff
+621 -593
View File
File diff suppressed because it is too large Load Diff
+35 -13
View File
@@ -61,22 +61,44 @@ export abstract class HyperionWorker {
}).catch(console.log);
});
const bytesToString = (bytes: number) => {
const e = Math.log(bytes) / Math.log(1024) | 0;
const n = (bytes / Math.pow(1024, e)).toFixed(2);
return n + ' ' + (e == 0 ? 'bytes' : (['KB', 'MB', 'GB', 'TB'])[e - 1]);
};
// handle ipc messages
process.on('message', (msg: any) => {
if (msg.event === 'request_v8_heap_stats') {
const report: HeapInfo = v8.getHeapStatistics();
const used_pct = report.used_heap_size / report.heap_size_limit;
process.send({
event: 'v8_heap_report',
id: process.env.worker_role + ':' + process.env.worker_id,
data: {
heap_usage: (used_pct * 100).toFixed(2) + "%",
...report
}
});
return;
switch (msg.event) {
case 'request_v8_heap_stats': {
const report: HeapInfo = v8.getHeapStatistics();
const used_pct = report.used_heap_size / report.heap_size_limit;
process.send({
event: 'v8_heap_report',
id: process.env.worker_role + ':' + process.env.worker_id,
data: {
heap_usage: (used_pct * 100).toFixed(2) + "%",
...report
}
});
break;
}
case 'request_memory_usage': {
const report = process.memoryUsage();
process.send({
event: 'memory_report',
id: process.env.worker_role + ':' + process.env.worker_id,
data: {
resident: bytesToString(report.rss),
}
});
break;
}
default: {
this.onIpcMessage(msg);
}
}
this.onIpcMessage(msg);
});
}
+7 -6
View File
@@ -291,7 +291,7 @@ export default class StateReader extends HyperionWorker {
return true;
}
private async onMessage(data: any) {
private async onMessage(data: Buffer) {
// this.tempBlockSizeSum += data.length;
if (!this.abi) {
@@ -472,13 +472,14 @@ export default class StateReader extends HyperionWorker {
}
}
private processFirstABI(data: any) {
AbiEOS.load_abi("0", data);
this.abi = JSON.parse(data);
private processFirstABI(data: Buffer) {
const abiString = data.toString();
AbiEOS.load_abi("0", abiString);
this.abi = JSON.parse(abiString);
this.types = Serialize.getTypesFromAbi(Serialize.createInitialTypes(), this.abi);
this.abi.tables.map(table => this.tables.set(table.name, table.type));
// notify master about first abi
process.send({event: 'init_abi', data: data});
process.send({event: 'init_abi', data: abiString});
// request status
this.send(['get_status_request_v0', {}]);
}
@@ -545,7 +546,7 @@ export default class StateReader extends HyperionWorker {
private async logForkEvent(starting_block, ending_block, new_id) {
process.send({event: 'fork_event', data: {starting_block, ending_block, new_id}});
await this.client.index({
index: this.chain + '-logs',
index: this.chain + '-logs-' + this.conf.settings.index_version,
body: {
type: 'fork',
'@timestamp': new Date().toISOString(),