machankura/mempool
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Import LN historical statistics (network wide + per node)

Browse Source
This commit is contained in:
nymkappa 2022-08-01 17:25:44 +02:00
parent 9126387847
commit 068f7392dd
7 changed files with 440 additions and 183 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
backend/package-lock.json generated
View File

@ -31,6 +31,7 @@
"@typescript-eslint/parser": "^5.30.5",
"eslint": "^8.19.0",
"eslint-config-prettier": "^8.5.0",
"fast-xml-parser": "^4.0.9",
"prettier": "^2.7.1"
}
},
@ -1496,6 +1497,22 @@
"integrity": "sha512-DCXu6Ifhqcks7TZKY3Hxp3y6qphY5SJZmrWMDrKcERSOXWQdMhU9Ig/PYrzyw/ul9jOIyh0N4M0tbC5hodg8dw==",
"dev": true
},
"node_modules/fast-xml-parser": {
"version": "4.0.9",
"resolved": "https://registry.npmjs.org/fast-xml-parser/-/fast-xml-parser-4.0.9.tgz",
"integrity": "sha512-4G8EzDg2Nb1Qurs3f7BpFV4+jpMVsdgLVuG1Uv8O2OHJfVCg7gcA53obuKbmVqzd4Y7YXVBK05oJG7hzGIdyzg==",
"dev": true,
"dependencies": {
"strnum": "^1.0.5"
},
"bin": {
"fxparser": "src/cli/cli.js"
},
"funding": {
"type": "paypal",
"url": "https://paypal.me/naturalintelligence"
}
},
"node_modules/fastq": {
"version": "1.13.0",
"resolved": "https://registry.npmjs.org/fastq/-/fastq-1.13.0.tgz",
@ -2665,6 +2682,12 @@
"url": "https://github.com/sponsors/sindresorhus"
}
},
"node_modules/strnum": {
"version": "1.0.5",
"resolved": "https://registry.npmjs.org/strnum/-/strnum-1.0.5.tgz",
"integrity": "sha512-J8bbNyKKXl5qYcR36TIO8W3mVGVHrmmxsd5PAItGkmyzwJvybiw2IVq5nqd0i4LSNSkB/sx9VHllbfFdr9k1JA==",
"dev": true
},
"node_modules/text-table": {
"version": "0.2.0",
"resolved": "https://registry.npmjs.org/text-table/-/text-table-0.2.0.tgz",
@ -3973,6 +3996,15 @@
"integrity": "sha512-DCXu6Ifhqcks7TZKY3Hxp3y6qphY5SJZmrWMDrKcERSOXWQdMhU9Ig/PYrzyw/ul9jOIyh0N4M0tbC5hodg8dw==",
"dev": true
},
"fast-xml-parser": {
"version": "4.0.9",
"resolved": "https://registry.npmjs.org/fast-xml-parser/-/fast-xml-parser-4.0.9.tgz",
"integrity": "sha512-4G8EzDg2Nb1Qurs3f7BpFV4+jpMVsdgLVuG1Uv8O2OHJfVCg7gcA53obuKbmVqzd4Y7YXVBK05oJG7hzGIdyzg==",
"dev": true,
"requires": {
"strnum": "^1.0.5"
}
},
"fastq": {
"version": "1.13.0",
"resolved": "https://registry.npmjs.org/fastq/-/fastq-1.13.0.tgz",
@ -4817,6 +4849,12 @@
"integrity": "sha512-6fPc+R4ihwqP6N/aIv2f1gMH8lOVtWQHoqC4yK6oSDVVocumAsfCqjkXnqiYMhmMwS/mEHLp7Vehlt3ql6lEig==",
"dev": true
},
"strnum": {
"version": "1.0.5",
"resolved": "https://registry.npmjs.org/strnum/-/strnum-1.0.5.tgz",
"integrity": "sha512-J8bbNyKKXl5qYcR36TIO8W3mVGVHrmmxsd5PAItGkmyzwJvybiw2IVq5nqd0i4LSNSkB/sx9VHllbfFdr9k1JA==",
"dev": true
},
"text-table": {
"version": "0.2.0",
"resolved": "https://registry.npmjs.org/text-table/-/text-table-0.2.0.tgz",

1
backend/package.json
View File

@ -53,6 +53,7 @@
"@typescript-eslint/parser": "^5.30.5",
"eslint": "^8.19.0",
"eslint-config-prettier": "^8.5.0",
"fast-xml-parser": "^4.0.9",
"prettier": "^2.7.1"
}
}

6
backend/src/api/database-migration.ts
View File

@ -4,7 +4,7 @@ import logger from '../logger';
import { Common } from './common';
class DatabaseMigration {
private static currentVersion = 33;
private static currentVersion = 34;
private queryTimeout = 120000;
private statisticsAddedIndexed = false;
private uniqueLogs: string[] = [];
@ -311,6 +311,10 @@ class DatabaseMigration {
if (databaseSchemaVersion < 33 && isBitcoin == true) {
await this.$executeQuery('ALTER TABLE `geo_names` CHANGE `type` `type` enum("city","country","division","continent","as_organization", "country_iso_code") NOT NULL');
}
if (databaseSchemaVersion < 34 && isBitcoin == true) {
await this.$executeQuery('ALTER TABLE `lightning_stats` ADD clearnet_tor_nodes int(11) NOT NULL DEFAULT "0"');
}
}
/**

4
backend/src/config.ts
View File

@ -31,6 +31,7 @@ interface IConfig {
LIGHTNING: {
ENABLED: boolean;
BACKEND: 'lnd' | 'cln' | 'ldk';
TOPOLOGY_FOLDER: string;
};
LND: {
TLS_CERT_PATH: string;
@ -177,7 +178,8 @@ const defaults: IConfig = {
},
'LIGHTNING': {
'ENABLED': false,
'BACKEND': 'lnd'
'BACKEND': 'lnd',
'TOPOLOGY_FOLDER': '',
},
'LND': {
'TLS_CERT_PATH': '',

183
backend/src/tasks/lightning/stats-updater.service.ts
View File

@ -3,7 +3,7 @@ import DB from '../../database';
import logger from '../../logger';
import lightningApi from '../../api/lightning/lightning-api-factory';
import channelsApi from '../../api/explorer/channels.api';
import * as net from 'net';
import { isIP } from 'net';
class LightningStatsUpdater {
hardCodedStartTime = '2018-01-12';
@ -28,9 +28,6 @@ class LightningStatsUpdater {
return;
}
await this.$populateHistoricalStatistics();
await this.$populateHistoricalNodeStatistics();
setTimeout(() => {
this.$runTasks();
}, this.timeUntilMidnight());
@ -85,7 +82,7 @@ class LightningStatsUpdater {
if (hasOnion) {
torNodes++;
}
const hasClearnet = [4, 6].includes(net.isIP(socket.addr.split(':')[0]));
const hasClearnet = [4, 6].includes(isIP(socket.split(':')[0]));
if (hasClearnet) {
clearnetNodes++;
}
@ -167,182 +164,6 @@ class LightningStatsUpdater {
logger.err('$logNodeStatsDaily() error: ' + (e instanceof Error ? e.message : e));
}
}
// We only run this on first launch
private async $populateHistoricalStatistics() {
try {
const [rows]: any = await DB.query(`SELECT COUNT(*) FROM lightning_stats`);
// Only run if table is empty
if (rows[0]['COUNT(*)'] > 0) {
return;
}
logger.info(`Running historical stats population...`);
const [channels]: any = await DB.query(`SELECT capacity, created, closing_date FROM channels ORDER BY created ASC`);
const [nodes]: any = await DB.query(`SELECT first_seen, sockets FROM nodes ORDER BY first_seen ASC`);
const date: Date = new Date(this.hardCodedStartTime);
const currentDate = new Date();
this.setDateMidnight(currentDate);
while (date < currentDate) {
let totalCapacity = 0;
let channelsCount = 0;
for (const channel of channels) {
if (new Date(channel.created) > date) {
break;
}
if (channel.closing_date === null || new Date(channel.closing_date) > date) {
totalCapacity += channel.capacity;
channelsCount++;
}
}
let nodeCount = 0;
let clearnetNodes = 0;
let torNodes = 0;
let unannouncedNodes = 0;
for (const node of nodes) {
if (new Date(node.first_seen) > date) {
break;
}
nodeCount++;
const sockets = node.sockets.split(',');
let isUnnanounced = true;
for (const socket of sockets) {
const hasOnion = socket.indexOf('.onion') !== -1;
if (hasOnion) {
torNodes++;
isUnnanounced = false;
}
const hasClearnet = [4, 6].includes(net.isIP(socket.substring(0, socket.lastIndexOf(':'))));
if (hasClearnet) {
clearnetNodes++;
isUnnanounced = false;
}
}
if (isUnnanounced) {
unannouncedNodes++;
}
}
const query = `INSERT INTO lightning_stats(
added,
channel_count,
node_count,
total_capacity,
tor_nodes,
clearnet_nodes,
unannounced_nodes,
avg_capacity,
avg_fee_rate,
avg_base_fee_mtokens,
med_capacity,
med_fee_rate,
med_base_fee_mtokens
)
VALUES (FROM_UNIXTIME(?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`;
const rowTimestamp = date.getTime() / 1000; // Save timestamp for the row insertion down below
date.setUTCDate(date.getUTCDate() + 1);
// Last iteration, save channels stats
const channelStats = (date >= currentDate ? await channelsApi.$getChannelsStats() : undefined);
await DB.query(query, [
rowTimestamp,
channelsCount,
nodeCount,
totalCapacity,
torNodes,
clearnetNodes,
unannouncedNodes,
channelStats?.avgCapacity ?? 0,
channelStats?.avgFeeRate ?? 0,
channelStats?.avgBaseFee ?? 0,
channelStats?.medianCapacity ?? 0,
channelStats?.medianFeeRate ?? 0,
channelStats?.medianBaseFee ?? 0,
]);
}
logger.info('Historical stats populated.');
} catch (e) {
logger.err('$populateHistoricalData() error: ' + (e instanceof Error ? e.message : e));
}
}
private async $populateHistoricalNodeStatistics() {
try {
const [rows]: any = await DB.query(`SELECT COUNT(*) FROM node_stats`);
// Only run if table is empty
if (rows[0]['COUNT(*)'] > 0) {
return;
}
logger.info(`Running historical node stats population...`);
const [nodes]: any = await DB.query(`SELECT public_key, first_seen, alias FROM nodes ORDER BY first_seen ASC`);
for (const node of nodes) {
const [channels]: any = await DB.query(`SELECT capacity, created, closing_date FROM channels WHERE node1_public_key = ? OR node2_public_key = ? ORDER BY created ASC`, [node.public_key, node.public_key]);
const date: Date = new Date(this.hardCodedStartTime);
const currentDate = new Date();
this.setDateMidnight(currentDate);
let lastTotalCapacity = 0;
let lastChannelsCount = 0;
while (date < currentDate) {
let totalCapacity = 0;
let channelsCount = 0;
for (const channel of channels) {
if (new Date(channel.created) > date) {
break;
}
if (channel.closing_date !== null && new Date(channel.closing_date) < date) {
date.setUTCDate(date.getUTCDate() + 1);
continue;
}
totalCapacity += channel.capacity;
channelsCount++;
}
if (lastTotalCapacity === totalCapacity && lastChannelsCount === channelsCount) {
date.setUTCDate(date.getUTCDate() + 1);
continue;
}
lastTotalCapacity = totalCapacity;
lastChannelsCount = channelsCount;
const query = `INSERT INTO node_stats(
public_key,
added,
capacity,
channels
)
VALUES (?, FROM_UNIXTIME(?), ?, ?)`;
await DB.query(query, [
node.public_key,
date.getTime() / 1000,
totalCapacity,
channelsCount,
]);
date.setUTCDate(date.getUTCDate() + 1);
}
logger.debug('Updated node_stats for: ' + node.alias);
}
logger.info('Historical stats populated.');
} catch (e) {
logger.err('$populateHistoricalNodeData() error: ' + (e instanceof Error ? e.message : e));
}
}
}
export default new LightningStatsUpdater();

104
backend/src/tasks/lightning/sync-tasks/funding-tx-fetcher.ts Normal file
View File

@ -0,0 +1,104 @@
import { existsSync, readFileSync, writeFileSync } from 'fs';
import bitcoinClient from '../../../api/bitcoin/bitcoin-client';
import config from '../../../config';
import logger from '../../../logger';
const BLOCKS_CACHE_MAX_SIZE = 100;
const CACHE_FILE_NAME = config.MEMPOOL.CACHE_DIR + '/ln-funding-txs-cache.json';
class FundingTxFetcher {
private running = false;
private blocksCache = {};
private channelNewlyProcessed = 0;
public fundingTxCache = {};
async $fetchChannelsFundingTxs(channelIds: string[]): Promise<void> {
if (this.running) {
return;
}
this.running = true;
// Load funding tx disk cache
if (Object.keys(this.fundingTxCache).length === 0 && existsSync(CACHE_FILE_NAME)) {
try {
this.fundingTxCache = JSON.parse(readFileSync(CACHE_FILE_NAME, 'utf-8'));
} catch (e) {
logger.err(`Unable to parse channels funding txs disk cache. Starting from scratch`);
this.fundingTxCache = {};
}
logger.debug(`Imported ${Object.keys(this.fundingTxCache).length} funding tx amount from the disk cache`);
}
const globalTimer = new Date().getTime() / 1000;
let cacheTimer = new Date().getTime() / 1000;
let loggerTimer = new Date().getTime() / 1000;
let channelProcessed = 0;
this.channelNewlyProcessed = 0;
for (const channelId of channelIds) {
await this.$fetchChannelOpenTx(channelId);
++channelProcessed;
let elapsedSeconds = Math.round((new Date().getTime() / 1000) - loggerTimer);
if (elapsedSeconds > 10) {
elapsedSeconds = Math.round((new Date().getTime() / 1000) - globalTimer);
logger.debug(`Indexing channels funding tx ${channelProcessed + 1} of ${channelIds.length} ` +
`(${Math.floor(channelProcessed / channelIds.length * 10000) / 100}%) | ` +
`elapsed: ${elapsedSeconds} seconds`
);
loggerTimer = new Date().getTime() / 1000;
}
elapsedSeconds = Math.round((new Date().getTime() / 1000) - cacheTimer);
if (elapsedSeconds > 60) {
logger.debug(`Saving ${Object.keys(this.fundingTxCache).length} funding txs cache into disk`);
writeFileSync(CACHE_FILE_NAME, JSON.stringify(this.fundingTxCache));
cacheTimer = new Date().getTime() / 1000;
}
}
if (this.channelNewlyProcessed > 0) {
logger.info(`Indexed ${this.channelNewlyProcessed} additional channels funding tx`);
logger.debug(`Saving ${Object.keys(this.fundingTxCache).length} funding txs cache into disk`);
writeFileSync(CACHE_FILE_NAME, JSON.stringify(this.fundingTxCache));
}
this.running = false;
}
public async $fetchChannelOpenTx(channelId: string): Promise<any> {
if (this.fundingTxCache[channelId]) {
return this.fundingTxCache[channelId];
}
const parts = channelId.split('x');
const blockHeight = parts[0];
const txIdx = parts[1];
const outputIdx = parts[2];
let block = this.blocksCache[blockHeight];
if (!block) {
const blockHash = await bitcoinClient.getBlockHash(parseInt(blockHeight, 10));
block = await bitcoinClient.getBlock(blockHash, 2);
this.blocksCache[block.height] = block;
}
const blocksCacheHashes = Object.keys(this.blocksCache).sort();
if (blocksCacheHashes.length > BLOCKS_CACHE_MAX_SIZE) {
for (let i = 0; i < 10; ++i) {
delete this.blocksCache[blocksCacheHashes[i]];
}
}
this.fundingTxCache[channelId] = {
timestamp: block.time,
txid: block.tx[txIdx].txid,
value: block.tx[txIdx].vout[outputIdx].value,
};
++this.channelNewlyProcessed;
return this.fundingTxCache[channelId];
}
}
export default new FundingTxFetcher;

287
backend/src/tasks/lightning/sync-tasks/stats-importer.ts Normal file
View File

@ -0,0 +1,287 @@
import DB from '../../../database';
import { readdirSync, readFileSync } from 'fs';
import { XMLParser } from 'fast-xml-parser';
import logger from '../../../logger';
import fundingTxFetcher from './funding-tx-fetcher';
import config from '../../../config';
interface Node {
id: string;
timestamp: number;
features: string;
rgb_color: string;
alias: string;
addresses: string;
out_degree: number;
in_degree: number;
}
interface Channel {
scid: string;
source: string;
destination: string;
timestamp: number;
features: string;
fee_base_msat: number;
fee_proportional_millionths: number;
htlc_minimim_msat: number;
cltv_expiry_delta: number;
htlc_maximum_msat: number;
}
const topologiesFolder = config.LIGHTNING.TOPOLOGY_FOLDER;
const parser = new XMLParser();
let latestNodeCount = 1; // Ignore gap in the data
async function $run(): Promise<void> {
// const [channels]: any[] = await DB.query('SELECT short_id from channels;');
// logger.info('Caching funding txs for currently existing channels');
// await fundingTxFetcher.$fetchChannelsFundingTxs(channels.map(channel => channel.short_id));
await $importHistoricalLightningStats();
}
/**
* Parse the file content into XML, and return a list of nodes and channels
*/
function parseFile(fileContent): any {
const graph = parser.parse(fileContent);
if (Object.keys(graph).length === 0) {
return null;
}
const nodes: Node[] = [];
const channels: Channel[] = [];
// If there is only one entry, the parser does not return an array, so we override this
if (!Array.isArray(graph.graphml.graph.node)) {
graph.graphml.graph.node = [graph.graphml.graph.node];
}
if (!Array.isArray(graph.graphml.graph.edge)) {
graph.graphml.graph.edge = [graph.graphml.graph.edge];
}
for (const node of graph.graphml.graph.node) {
if (!node.data) {
continue;
}
nodes.push({
id: node.data[0],
timestamp: node.data[1],
features: node.data[2],
rgb_color: node.data[3],
alias: node.data[4],
addresses: node.data[5],
out_degree: node.data[6],
in_degree: node.data[7],
});
}
for (const channel of graph.graphml.graph.edge) {
if (!channel.data) {
continue;
}
channels.push({
scid: channel.data[0],
source: channel.data[1],
destination: channel.data[2],
timestamp: channel.data[3],
features: channel.data[4],
fee_base_msat: channel.data[5],
fee_proportional_millionths: channel.data[6],
htlc_minimim_msat: channel.data[7],
cltv_expiry_delta: channel.data[8],
htlc_maximum_msat: channel.data[9],
});
}
return {
nodes: nodes,
channels: channels,
};
}
/**
* Generate LN network stats for one day
*/
async function computeNetworkStats(timestamp: number, networkGraph): Promise<void> {
// Node counts and network shares
let clearnetNodes = 0;
let torNodes = 0;
let clearnetTorNodes = 0;
let unannouncedNodes = 0;
for (const node of networkGraph.nodes) {
let hasOnion = false;
let hasClearnet = false;
let isUnnanounced = true;
const sockets = node.addresses.split(',');
for (const socket of sockets) {
hasOnion = hasOnion || (socket.indexOf('torv3://') !== -1);
hasClearnet = hasClearnet || (socket.indexOf('ipv4://') !== -1 || socket.indexOf('ipv6://') !== -1);
}
if (hasOnion && hasClearnet) {
clearnetTorNodes++;
isUnnanounced = false;
} else if (hasOnion) {
torNodes++;
isUnnanounced = false;
} else if (hasClearnet) {
clearnetNodes++;
isUnnanounced = false;
}
if (isUnnanounced) {
unannouncedNodes++;
}
}
// Channels and node historical stats
const nodeStats = {};
let capacity = 0;
let avgFeeRate = 0;
let avgBaseFee = 0;
const capacities: number[] = [];
const feeRates: number[] = [];
const baseFees: number[] = [];
for (const channel of networkGraph.channels) {
const tx = await fundingTxFetcher.$fetchChannelOpenTx(channel.scid.slice(0, -2));
if (!tx) {
logger.err(`Unable to fetch funding tx for channel ${channel.scid}. Capacity and creation date will stay unknown.`);
continue;
}
if (!nodeStats[channel.source]) {
nodeStats[channel.source] = {
capacity: 0,
channels: 0,
};
}
if (!nodeStats[channel.destination]) {
nodeStats[channel.destination] = {
capacity: 0,
channels: 0,
};
}
nodeStats[channel.source].capacity += Math.round(tx.value * 100000000);
nodeStats[channel.source].channels++;
nodeStats[channel.destination].capacity += Math.round(tx.value * 100000000);
nodeStats[channel.destination].channels++;
capacity += Math.round(tx.value * 100000000);
avgFeeRate += channel.fee_proportional_millionths;
avgBaseFee += channel.fee_base_msat;
capacities.push(Math.round(tx.value * 100000000));
feeRates.push(channel.fee_proportional_millionths);
baseFees.push(channel.fee_base_msat);
}
avgFeeRate /= networkGraph.channels.length;
avgBaseFee /= networkGraph.channels.length;
const medCapacity = capacities.sort((a, b) => b - a)[Math.round(capacities.length / 2 - 1)];
const medFeeRate = feeRates.sort((a, b) => b - a)[Math.round(feeRates.length / 2 - 1)];
const medBaseFee = baseFees.sort((a, b) => b - a)[Math.round(baseFees.length / 2 - 1)];
let query = `INSERT INTO lightning_stats(
added,
channel_count,
node_count,
total_capacity,
tor_nodes,
clearnet_nodes,
unannounced_nodes,
clearnet_tor_nodes,
avg_capacity,
avg_fee_rate,
avg_base_fee_mtokens,
med_capacity,
med_fee_rate,
med_base_fee_mtokens
)
VALUES (FROM_UNIXTIME(?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`;
await DB.query(query, [
timestamp,
networkGraph.channels.length,
networkGraph.nodes.length,
capacity,
torNodes,
clearnetNodes,
unannouncedNodes,
clearnetTorNodes,
Math.round(capacity / networkGraph.channels.length),
avgFeeRate,
avgBaseFee,
medCapacity,
medFeeRate,
medBaseFee,
]);
for (const public_key of Object.keys(nodeStats)) {
query = `INSERT INTO node_stats(
public_key,
added,
capacity,
channels
)
VALUES (?, FROM_UNIXTIME(?), ?, ?)`;
await DB.query(query, [
public_key,
timestamp,
nodeStats[public_key].capacity,
nodeStats[public_key].channels,
]);
}
}
export async function $importHistoricalLightningStats(): Promise<void> {
const fileList = readdirSync(topologiesFolder);
fileList.sort().reverse();
const [rows]: any[] = await DB.query('SELECT UNIX_TIMESTAMP(added) as added FROM lightning_stats');
const existingStatsTimestamps = {};
for (const row of rows) {
existingStatsTimestamps[row.added] = true;
}
for (const filename of fileList) {
const timestamp = parseInt(filename.split('_')[1], 10);
const fileContent = readFileSync(`${topologiesFolder}/${filename}`, 'utf8');
const graph = parseFile(fileContent);
if (!graph) {
continue;
}
// Ignore drop of more than 90% of the node count as it's probably a missing data point
const diffRatio = graph.nodes.length / latestNodeCount;
if (diffRatio < 0.90) {
continue;
}
latestNodeCount = graph.nodes.length;
// Stats exist already, don't calculate/insert them
if (existingStatsTimestamps[timestamp] === true) {
continue;
}
logger.debug(`Processing ${topologiesFolder}/${filename}`);
const datestr = `${new Date(timestamp * 1000).toUTCString()} (${timestamp})`;
logger.debug(`${datestr}: Found ${graph.nodes.length} nodes and ${graph.channels.length} channels`);
// Cache funding txs
logger.debug(`Caching funding txs for ${datestr}`);
await fundingTxFetcher.$fetchChannelsFundingTxs(graph.channels.map(channel => channel.scid.slice(0, -2)));
logger.debug(`Generating LN network stats for ${datestr}`);
await computeNetworkStats(timestamp, graph);
}
logger.info(`Lightning network stats historical import completed`);
}
$run().then(() => process.exit(0));