wait for services to be available on indexer start
This commit is contained in:
Igor Lins e Silva
+208
-192
@@ -6,245 +6,261 @@ import {join} from "path";
|
||||
let config;
|
||||
const conf_path = join(__dirname, `../${process.env.CONFIG_JSON}`);
|
||||
if (existsSync(conf_path)) {
|
||||
try {
|
||||
config = require(conf_path);
|
||||
} catch (e) {
|
||||
console.log(e.message);
|
||||
process.exit(1);
|
||||
}
|
||||
try {
|
||||
config = require(conf_path);
|
||||
} catch (e) {
|
||||
console.log(e.message);
|
||||
process.exit(1);
|
||||
}
|
||||
}
|
||||
|
||||
if (!config) {
|
||||
console.log(`Configuration not found: ${conf_path}`);
|
||||
process.exit(1);
|
||||
console.log(`Configuration not found: ${conf_path}`);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
function getLastResult(results: ApiResponse) {
|
||||
if (results.body.hits?.hits?.length > 0) {
|
||||
return parseInt(results.body.hits.hits[0].sort[0], 10);
|
||||
} else {
|
||||
return 1;
|
||||
}
|
||||
if (results.body.hits?.hits?.length > 0) {
|
||||
return parseInt(results.body.hits.hits[0].sort[0], 10);
|
||||
} else {
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
export async function getLastIndexedBlockByDelta(es_client: Client, chain: string) {
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-delta-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {bool: {filter: {match_all: {}}}},
|
||||
sort: [{block_num: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-delta-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {bool: {filter: {match_all: {}}}},
|
||||
sort: [{block_num: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
}
|
||||
|
||||
export async function getLastIndexedBlock(es_client: Client, chain: string) {
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-block-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {bool: {filter: {match_all: {}}}},
|
||||
sort: [{block_num: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-block-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {bool: {filter: {match_all: {}}}},
|
||||
sort: [{block_num: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
}
|
||||
|
||||
export async function getLastIndexedBlockWithTotalBlocks(es_client: Client, chain: string): Promise<[number, number]> {
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-block-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {bool: {filter: {match_all: {}}}},
|
||||
sort: [{block_num: {order: "desc"}}],
|
||||
track_total_hits: true
|
||||
}
|
||||
});
|
||||
let lastBlock = getLastResult(results);
|
||||
let totalBlocks = results.body.hits.total.value || 1;
|
||||
return [lastBlock, totalBlocks];
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-block-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {bool: {filter: {match_all: {}}}},
|
||||
sort: [{block_num: {order: "desc"}}],
|
||||
track_total_hits: true
|
||||
}
|
||||
});
|
||||
let lastBlock = getLastResult(results);
|
||||
let totalBlocks = results.body.hits.total.value || 1;
|
||||
return [lastBlock, totalBlocks];
|
||||
}
|
||||
|
||||
export async function getFirstIndexedBlock(es_client: Client, chain: string): Promise<number> {
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-block-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {bool: {filter: {match_all: {}}}},
|
||||
sort: [{block_num: {order: "asc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-block-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {bool: {filter: {match_all: {}}}},
|
||||
sort: [{block_num: {order: "asc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
}
|
||||
|
||||
|
||||
export async function getLastIndexedABI(es_client: Client, chain: string) {
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-abi-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {
|
||||
match_all: {}
|
||||
},
|
||||
sort: [{block: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-abi-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {
|
||||
match_all: {}
|
||||
},
|
||||
sort: [{block: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
}
|
||||
|
||||
export async function getLastIndexedBlockByDeltaFromRange(es_client: Client, chain: string, first: number, last: number) {
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-delta-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {
|
||||
range: {
|
||||
block_num: {
|
||||
"gte": first,
|
||||
"lt": last,
|
||||
"boost": 2
|
||||
}
|
||||
}
|
||||
},
|
||||
sort: [{block_num: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
const results: ApiResponse = await es_client.search({
|
||||
index: chain + '-delta-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {
|
||||
range: {
|
||||
block_num: {
|
||||
"gte": first,
|
||||
"lt": last,
|
||||
"boost": 2
|
||||
}
|
||||
}
|
||||
},
|
||||
sort: [{block_num: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
}
|
||||
|
||||
export async function getLastIndexedBlockFromRange(es_client: Client, chain: string, first: number, last: number) {
|
||||
const results = await es_client.search({
|
||||
index: chain + '-block-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {
|
||||
range: {
|
||||
block_num: {
|
||||
"gte": first,
|
||||
"lt": last,
|
||||
"boost": 2
|
||||
}
|
||||
}
|
||||
},
|
||||
sort: [{block_num: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
const results = await es_client.search({
|
||||
index: chain + '-block-*',
|
||||
size: 1,
|
||||
body: {
|
||||
query: {
|
||||
range: {
|
||||
block_num: {
|
||||
"gte": first,
|
||||
"lt": last,
|
||||
"boost": 2
|
||||
}
|
||||
}
|
||||
},
|
||||
sort: [{block_num: {order: "desc"}}]
|
||||
}
|
||||
});
|
||||
return getLastResult(results);
|
||||
}
|
||||
|
||||
export function messageAllWorkers(cl, payload) {
|
||||
for (const c in cl.workers) {
|
||||
if (cl.workers.hasOwnProperty(c)) {
|
||||
const _w = cl.workers[c];
|
||||
if (_w) {
|
||||
try {
|
||||
if (_w.isConnected()) {
|
||||
_w.send(payload);
|
||||
} else {
|
||||
hLog('Worker is not connected!');
|
||||
}
|
||||
} catch (e) {
|
||||
hLog('Failed to message worker!');
|
||||
hLog(e);
|
||||
}
|
||||
} else {
|
||||
hLog('Worker not found!');
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const c in cl.workers) {
|
||||
if (cl.workers.hasOwnProperty(c)) {
|
||||
const _w = cl.workers[c];
|
||||
if (_w) {
|
||||
try {
|
||||
if (_w.isConnected()) {
|
||||
_w.send(payload);
|
||||
} else {
|
||||
hLog('Worker is not connected!');
|
||||
}
|
||||
} catch (e) {
|
||||
hLog('Failed to message worker!');
|
||||
hLog(e);
|
||||
}
|
||||
} else {
|
||||
hLog('Worker not found!');
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function serialize(type, value, txtEnc, txtDec, types) {
|
||||
const buffer = new Serialize.SerialBuffer({
|
||||
textEncoder: txtEnc,
|
||||
textDecoder: txtDec
|
||||
});
|
||||
Serialize.getType(types, type).serialize(buffer, value);
|
||||
return buffer.asUint8Array();
|
||||
const buffer = new Serialize.SerialBuffer({
|
||||
textEncoder: txtEnc,
|
||||
textDecoder: txtDec
|
||||
});
|
||||
Serialize.getType(types, type).serialize(buffer, value);
|
||||
return buffer.asUint8Array();
|
||||
}
|
||||
|
||||
export function deserialize(type, array, txtEnc, txtDec, types) {
|
||||
const buffer = new Serialize.SerialBuffer({
|
||||
textEncoder: txtEnc,
|
||||
textDecoder: txtDec,
|
||||
array
|
||||
});
|
||||
return Serialize.getType(types, type).deserialize(buffer, new Serialize.SerializerState({bytesAsUint8Array: true}));
|
||||
const buffer = new Serialize.SerialBuffer({
|
||||
textEncoder: txtEnc,
|
||||
textDecoder: txtDec,
|
||||
array
|
||||
});
|
||||
return Serialize.getType(types, type).deserialize(buffer, new Serialize.SerializerState({bytesAsUint8Array: true}));
|
||||
}
|
||||
|
||||
function getNested(path_array, jsonObj) {
|
||||
const nextPath = path_array.shift();
|
||||
const nextValue = jsonObj[nextPath];
|
||||
if (!nextValue) {
|
||||
return null;
|
||||
} else {
|
||||
if (typeof nextValue !== 'object') {
|
||||
return nextValue;
|
||||
} else {
|
||||
if (Array.isArray(nextValue)) {
|
||||
return nextValue;
|
||||
} else {
|
||||
return getNested(path_array, nextValue);
|
||||
}
|
||||
}
|
||||
}
|
||||
const nextPath = path_array.shift();
|
||||
const nextValue = jsonObj[nextPath];
|
||||
if (!nextValue) {
|
||||
return null;
|
||||
} else {
|
||||
if (typeof nextValue !== 'object') {
|
||||
return nextValue;
|
||||
} else {
|
||||
if (Array.isArray(nextValue)) {
|
||||
return nextValue;
|
||||
} else {
|
||||
return getNested(path_array, nextValue);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function checkFilter(filter, _source) {
|
||||
if (filter.field && filter.value) {
|
||||
let fieldValue = getNested(filter.field.split("."), _source);
|
||||
if (!fieldValue) {
|
||||
const fArray = filter.field.split(".");
|
||||
if (fArray[0].startsWith('@')) {
|
||||
const actName = fArray[0].replace('@', '');
|
||||
if (_source.act.name === actName) {
|
||||
fArray[0] = 'data';
|
||||
fArray.unshift('act');
|
||||
fieldValue = getNested(fArray, _source);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (fieldValue) {
|
||||
if (Array.isArray(fieldValue)) {
|
||||
return fieldValue.indexOf(filter.value) !== -1;
|
||||
} else {
|
||||
return fieldValue === filter.value;
|
||||
}
|
||||
} else {
|
||||
return !filter.value;
|
||||
}
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
if (filter.field && filter.value) {
|
||||
let fieldValue = getNested(filter.field.split("."), _source);
|
||||
if (!fieldValue) {
|
||||
const fArray = filter.field.split(".");
|
||||
if (fArray[0].startsWith('@')) {
|
||||
const actName = fArray[0].replace('@', '');
|
||||
if (_source.act.name === actName) {
|
||||
fArray[0] = 'data';
|
||||
fArray.unshift('act');
|
||||
fieldValue = getNested(fArray, _source);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (fieldValue) {
|
||||
if (Array.isArray(fieldValue)) {
|
||||
return fieldValue.indexOf(filter.value) !== -1;
|
||||
} else {
|
||||
return fieldValue === filter.value;
|
||||
}
|
||||
} else {
|
||||
return !filter.value;
|
||||
}
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export function hLog(input: any, ...extra: any[]) {
|
||||
let role;
|
||||
if (process.env.worker_role) {
|
||||
const id = parseInt(process.env.worker_id);
|
||||
role = `[${process.pid} - ${(id < 10 ? '0' : '') + id.toString()}_${process.env.worker_role}]`;
|
||||
} else {
|
||||
if (process.env.script && process.env.script === './api/server.js') {
|
||||
role = `[${process.pid} - api]`;
|
||||
} else {
|
||||
role = `[${process.pid} - 00_master]`;
|
||||
}
|
||||
}
|
||||
if (process.env.TRACE_LOGS === 'true') {
|
||||
const e = new Error();
|
||||
const frame = e.stack.split("\n")[2];
|
||||
const where = frame.split(" ")[6].split(/[:()]/);
|
||||
const arr = where[1].split("/");
|
||||
const fileName = arr[arr.length - 1];
|
||||
const lineNumber = where[2];
|
||||
role += ` ${fileName}:${lineNumber}`;
|
||||
}
|
||||
console.log(role, input, ...extra);
|
||||
let role;
|
||||
if (process.env.worker_role) {
|
||||
const id = parseInt(process.env.worker_id);
|
||||
role = `[${process.pid} - ${(id < 10 ? '0' : '') + id.toString()}_${process.env.worker_role}]`;
|
||||
} else {
|
||||
if (process.env.script && process.env.script === './api/server.js') {
|
||||
role = `[${process.pid} - api]`;
|
||||
} else {
|
||||
role = `[${process.pid} - 00_master]`;
|
||||
}
|
||||
}
|
||||
if (process.env.TRACE_LOGS === 'true') {
|
||||
const e = new Error();
|
||||
const frame = e.stack.split("\n")[2];
|
||||
const where = frame.split(" ")[6].split(/[:()]/);
|
||||
const arr = where[1].split("/");
|
||||
const fileName = arr[arr.length - 1];
|
||||
const lineNumber = where[2];
|
||||
role += ` ${fileName}:${lineNumber}`;
|
||||
}
|
||||
console.log(role, input, ...extra);
|
||||
}
|
||||
|
||||
export function debugLog(text: any, ...extra: any[]) {
|
||||
if (config.settings.debug) {
|
||||
hLog(text, ...extra);
|
||||
}
|
||||
if (config.settings.debug) {
|
||||
hLog(text, ...extra);
|
||||
}
|
||||
}
|
||||
|
||||
export async function sleep(ms): Promise<void> {
|
||||
return new Promise(resolve => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
export async function waitUntilReady(executor: () => Promise<boolean>, attempts: number, interval: number, onError: () => void): Promise<void> {
|
||||
let i = 0;
|
||||
while (i < attempts) {
|
||||
if (await executor()) {
|
||||
return;
|
||||
}
|
||||
await sleep(interval);
|
||||
i++;
|
||||
}
|
||||
onError();
|
||||
}
|
||||
|
||||
+40
-22
@@ -12,7 +12,7 @@ import {
|
||||
getLastIndexedBlockByDeltaFromRange,
|
||||
getLastIndexedBlockFromRange,
|
||||
hLog,
|
||||
messageAllWorkers
|
||||
messageAllWorkers, waitUntilReady
|
||||
} from "../helpers/common_functions";
|
||||
|
||||
import {GetInfoResult} from "eosjs/dist/eosjs-rpc-interfaces";
|
||||
@@ -46,6 +46,7 @@ import {bootstrap} from 'global-agent';
|
||||
import moment = require("moment");
|
||||
import Timeout = NodeJS.Timeout;
|
||||
import {App, TemplatedApp, WebSocket} from "uWebSockets.js";
|
||||
import {checkQueueSize} from "../connections/amqp";
|
||||
|
||||
interface RevBlock {
|
||||
num: number;
|
||||
@@ -1721,43 +1722,60 @@ export class HyperionMaster {
|
||||
// Redis
|
||||
this.ioRedisClient = new IORedis(this.manager.conn.redis);
|
||||
|
||||
// Wait for Redis availability
|
||||
await waitUntilReady(async () => {
|
||||
try {
|
||||
await this.ioRedisClient.ping();
|
||||
return true;
|
||||
} catch (e) {
|
||||
return false;
|
||||
}
|
||||
}, 10, 5000, () => {
|
||||
hLog(`Redis not available, exiting...`);
|
||||
process.exit();
|
||||
});
|
||||
|
||||
// Remove first indexed block from cache (v2/health)
|
||||
await this.ioRedisClient.del(`${this.manager.chain}::fib`);
|
||||
|
||||
// wait for nodoes to be available, limit retries to 10 with a 5 seconds interval
|
||||
// sleep function
|
||||
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
|
||||
let retries = 0;
|
||||
while (true) {
|
||||
// Wait for Nodeos Chain API availability
|
||||
await waitUntilReady(async () => {
|
||||
try {
|
||||
const info = await this.rpc.get_info();
|
||||
if (info.server_version_string) {
|
||||
hLog(`Nodeos version: ${info.server_version_string}`);
|
||||
break;
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
} catch (e) {
|
||||
hLog(`Chain API Error: ${e.message}`);
|
||||
retries++;
|
||||
if (retries > 10) {
|
||||
hLog(`Chain API not available, exiting...`);
|
||||
process.exit();
|
||||
}
|
||||
await sleep(5000);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}, 10, 5000, () => {
|
||||
hLog(`Chain API not available, exiting...`);
|
||||
process.exit();
|
||||
});
|
||||
|
||||
// Elasticsearch
|
||||
// Wait for Elasticsearch availability
|
||||
this.client = this.manager.elasticsearchClient;
|
||||
try {
|
||||
const esInfo = await this.client.info();
|
||||
hLog(`Elasticsearch: ${esInfo.body.version.number} | Lucene: ${esInfo.body.version.lucene_version}`);
|
||||
this.emitAlert('info', `Indexer started using ES v${esInfo.body.version.number}`);
|
||||
} catch (e) {
|
||||
console.log(e.message);
|
||||
await waitUntilReady(async () => {
|
||||
try {
|
||||
const esInfo = await this.client.info();
|
||||
hLog(`Elasticsearch: ${esInfo.body.version.number} | Lucene: ${esInfo.body.version.lucene_version}`);
|
||||
this.emitAlert('info', `Indexer started using ES v${esInfo.body.version.number}`);
|
||||
return true;
|
||||
} catch (e) {
|
||||
console.log(e.message);
|
||||
return false;
|
||||
}
|
||||
}, 10, 5000, () => {
|
||||
hLog('Failed to check elasticsearch version!');
|
||||
process.exit();
|
||||
}
|
||||
});
|
||||
|
||||
await this.verifyIngestClients();
|
||||
|
||||
this.max_readers = this.conf.scaling.readers;
|
||||
if (this.conf.indexer.disable_reading) {
|
||||
this.max_readers = 1;
|
||||
|
||||
Reference in New Issue
Block a user