Compare commits

...

10 Commits

2 changed files with 128 additions and 26 deletions

View File

@ -15,7 +15,7 @@
"author": "Marco van Dijk", "author": "Marco van Dijk",
"license": "WTFPL", "license": "WTFPL",
"dependencies": { "dependencies": {
"ethers": "^6.8.1", "ethers": "5.7.2",
"express": "^4.17.1", "express": "^4.17.1",
"node-persist": "^3.1.3", "node-persist": "^3.1.3",
"npm": "^8.5.2", "npm": "^8.5.2",

View File

@ -18,7 +18,7 @@ const {
*/ */
const l1provider = new ethers.JsonRpcProvider( const l1provider = new ethers.providers.JsonRpcProvider(
CONF_API_L1_HTTP + CONF_API_L1_KEY CONF_API_L1_HTTP + CONF_API_L1_KEY
); );
const masterRouter = express.Router(); const masterRouter = express.Router();
@ -104,8 +104,7 @@ const getEnsDomain = async function (addr) {
return cached.domain ? cached.domain : cached.address; return cached.domain ? cached.domain : cached.address;
} }
// Refresh cause not cached or stale // Refresh cause not cached or stale
const address = ethers.getAddress(addr); const ensDomain = await l1provider.lookupAddress(addr);
const ensDomain = await l1provider.lookupAddress(address);
let ensObj; let ensObj;
if (!ensDomain) { if (!ensDomain) {
let domain = null; let domain = null;
@ -134,7 +133,18 @@ const getEnsDomain = async function (addr) {
); );
ensDomainCache[addr] = ensObj; ensDomainCache[addr] = ensObj;
await storage.setItem("ensDomainCache", ensDomainCache); await storage.setItem("ensDomainCache", ensDomainCache);
return ensObj.domain ? ensObj.domain : ensObj.address; if (ensObj.domain) {
// Update domain name
return ensObj.domain;
} else {
if (cached.domain) {
// Reuse last cached domain
return cached.domain;
} else {
// Return ETH addr
return ensObj.address;
}
}
} catch (err) { } catch (err) {
console.log(err); console.log(err);
console.log("Error looking up ENS info, retrying..."); console.log("Error looking up ENS info, retrying...");
@ -149,7 +159,7 @@ const getEnsDomain = async function (addr) {
*/ */
const updatePrometheus = async function (tag, id, instance, orchInfo) { const updatePrometheus = async function (tag, instance, orchInfo) {
const thisInstance = orchInfo.instances[instance]; const thisInstance = orchInfo.instances[instance];
const regionInfo = orchInfo.regionalStats[tag]; const regionInfo = orchInfo.regionalStats[tag];
if (regionInfo.latestDiscoveryTime) { if (regionInfo.latestDiscoveryTime) {
@ -200,13 +210,13 @@ const onOrchUpdate = async function (id, obj, tag, region, livepeer_regions) {
// Overwrite name with ENS domain if set // Overwrite name with ENS domain if set
let ensDomain = null; let ensDomain = null;
while (!ensDomain) { while (!ensDomain) {
ensDomain = await getEnsDomain(id); ensDomain = await getEnsDomain(id.toLowerCase());
} }
// Retrieve entry to update or init it // Retrieve entry to update or init it
let newObj = orchCache[id.toLowerCase()]; let newObj = orchCache[id.toLowerCase()];
if (!newObj) { if (!newObj) {
newObj = { newObj = {
name: obj.name, name: ensDomain,
regionalStats: {}, regionalStats: {},
instances: {}, instances: {},
leaderboardResults: { lastTime: now }, leaderboardResults: { lastTime: now },
@ -291,21 +301,48 @@ const onOrchUpdate = async function (id, obj, tag, region, livepeer_regions) {
} }
// Remove expired stuff // Remove expired stuff
for (const [id, obj] of Object.entries(newInstance.probedFrom)) { Object.keys(newInstance.probedFrom).forEach((key) => {
if (now - obj.lastTime > CONF_KEY_EXPIRY) { if (
newInstance.probedFrom[id] = null; !newInstance.probedFrom[key] ||
!newInstance.probedFrom[key].lastTime ||
now - newInstance.probedFrom[key].lastTime > CONF_KEY_EXPIRY
) {
console.log(
"Removing expired key " +
key +
" from the probed-from cache for orch " +
id
);
delete newInstance.probedFrom[key];
} }
} });
for (const [id, obj] of Object.entries(newInstance.regions)) { Object.keys(newInstance.regions).forEach((key) => {
if (now - obj.lastTime > CONF_KEY_EXPIRY) { if (
newInstance.regions[id] = null; !newInstance.regions[key] ||
!newInstance.regions[key].lastTime ||
now - newInstance.regions[key].lastTime > CONF_KEY_EXPIRY
) {
console.log(
"Removing expired key " + key + " from the regions cache for orch " + id
);
delete newInstance.regions[key];
} }
} });
for (const [id, obj] of Object.entries(newInstance.livepeer_regions)) { Object.keys(newInstance.livepeer_regions).forEach((key) => {
if (now - obj.lastTime > CONF_KEY_EXPIRY) { if (
newInstance.livepeer_regions[id] = null; !newInstance.livepeer_regions[key] ||
!newInstance.livepeer_regions[key].lastTime ||
now - newInstance.livepeer_regions[key].lastTime > CONF_KEY_EXPIRY
) {
console.log(
"Removing expired key " +
key +
" from the livepeer regions cache for orch " +
id
);
delete newInstance.livepeer_regions[key];
} }
} });
// Set last times for instance info // Set last times for instance info
newInstance.probedFrom[tag] = { newInstance.probedFrom[tag] = {
@ -331,9 +368,6 @@ const onOrchUpdate = async function (id, obj, tag, region, livepeer_regions) {
newInstance.longitude = obj.resolv.geoLookup.longitude; newInstance.longitude = obj.resolv.geoLookup.longitude;
} }
// Update name
newObj.name = obj.name;
// Finished updating // Finished updating
newObj.instances[obj.resolv.resolvedTarget] = newInstance; newObj.instances[obj.resolv.resolvedTarget] = newInstance;
newObj.regionalStats[tag] = newRegion; newObj.regionalStats[tag] = newRegion;
@ -341,8 +375,8 @@ const onOrchUpdate = async function (id, obj, tag, region, livepeer_regions) {
await storage.setItem("orchCache", orchCache); await storage.setItem("orchCache", orchCache);
// Update prometheus stats // Update prometheus stats
updatePrometheus(tag, id, obj.resolv.resolvedTarget, newObj); updatePrometheus(tag, obj.resolv.resolvedTarget, newObj);
console.log("Handled results for " + id + " from prober " + tag); console.log("Handled results for " + newObj.name + " from prober " + tag);
}; };
const updateCache = async function ( const updateCache = async function (
@ -436,7 +470,9 @@ const updateScore = async function (address) {
const newRTR = instance.round_trip_time / instance.seg_duration; const newRTR = instance.round_trip_time / instance.seg_duration;
let latitude = null; let latitude = null;
let longitude = null; let longitude = null;
for (const [resolvedTarget, instance] of Object.entries(thisInstances)) { for (const [resolvedTarget, instance] of Object.entries(
thisInstances
)) {
if (instance.livepeer_regions[region]) { if (instance.livepeer_regions[region]) {
latitude = instance.latitude; latitude = instance.latitude;
longitude = instance.longitude; longitude = instance.longitude;
@ -468,6 +504,18 @@ const updateScore = async function (address) {
}, },
newSR newSR
); );
if (
!orchCache[address.toLowerCase()].leaderboardResults[instance.region]
) {
orchCache[address.toLowerCase()].leaderboardResults[instance.region] =
{};
}
orchCache[address.toLowerCase()].leaderboardResults[
instance.region
].latestRTR = newRTR;
orchCache[address.toLowerCase()].leaderboardResults[
instance.region
].latestSR = newSR;
hasEdited = true; hasEdited = true;
} }
} }
@ -501,6 +549,60 @@ const recoverStorage = async function () {
if (storedOrchs) { if (storedOrchs) {
orchCache = storedOrchs; orchCache = storedOrchs;
} }
// Re-init from storage
for (const [id, obj] of Object.entries(orchCache)) {
const thisName = obj.name;
const thisInstances = obj.instances;
// Latest leaderboard results observed
if (obj.leaderboardResults) {
for (const [region, res] of Object.entries(obj.leaderboardResults)) {
// Skip the lastTime accessor - only use last observed regional stats
if (res.latestRTR == null || res.latestSR == null) {
continue;
}
console.log(
"Re-init leaderboard scores for orch=" +
id +
", RTR=" +
res.latestRTR +
" and success rate of " +
res.latestSR * 100 +
"%, livepeer region " +
region
);
let latitude = null;
let longitude = null;
for (const [resolvedTarget, instance] of Object.entries(
thisInstances
)) {
if (instance.livepeer_regions[region]) {
latitude = instance.latitude;
longitude = instance.longitude;
}
}
promLatestRTR.set(
{
livepeer_region: region,
orchestrator: thisName,
latitude: latitude,
longitude: longitude,
},
res.latestRTR
);
promLatestSuccessRate.set(
{
livepeer_region: region,
orchestrator: thisName,
latitude: latitude,
longitude: longitude,
},
res.latestSR
);
}
}
}
isSynced = true; isSynced = true;
}; };
recoverStorage(); recoverStorage();