Allow continuous execution (#1)

* remove location from env vars

* preparing for recurring execution

* rewrote and prepared for continuous execution

* added sample env

* fixing execution logs

* fixed log output

* update README with example startup commands

* make formatter happy

* fixed typo - thanks coderabbit

.env.sample

@@ -0,0 +1,8 @@
+UHTTP_LM_CLIENT_ID=
+UHTTP_LM_RPC_PROVIDER=https://gnosis.rpc-provider.prod.hoprnet.link
+UHTTP_LM_ZERO_HOP=true
+UHTTP_LM_DISCOVERY_PLATFORM=https://discovery.rpch.tech
+UHTTP_LM_INTERVAL_MS=60000
+UHTTP_LM_OFFSET_MS=11111
+UHTTP_LM_PUSH_GATEWAY=
+DEBUG=latency-monitor:*

README.md

@@ -1,3 +1,14 @@
 # u(nlinked)HTTP latency monitor
 
 Measure roundtrip latency through uHTTP network.
+
+## Development setup
+
+-   Copy `.env.sample` to `.env` and adjust values as needed.
+-   Build application with `yarn build`
+-   Start application with `export $(cat .env) && yarn start`
+
+## Docker setup
+
+-   Build container with `docker build -t uhttp-latency-monitor .`
+-   Run container with `docker run --env-file .env --platform linux/amd64 uhttp-latency-monitor`

package.json

@@ -9,7 +9,7 @@
         "prebuild": "node -p \"'export default \\'' + require('./package.json').version + '\\';'\" > src/version.ts",
         "build": "yarn prebuild && tsc",
         "format": "prettier --write src/ *.json *.md",
-        "format:ci": "prettier --check src/ *.js *.ts *.json *.md",
+        "format:ci": "prettier --check src/ *.json *.md",
         "lint": "eslint --fix .",
         "lint:ci": "eslint --max-warnings 0 .",
         "start": "node build/index.js",

src/index.ts

@@ -1,9 +1,18 @@
 import prom from 'prom-client';
+import { Routing } from '@hoprnet/uhttp-lib';
 
 import Version from './version';
 import log from './logger';
 import * as runner from './runner';
 
+type UHTTPsettings = Routing.Settings & { uClientId: string; rpcProvider: string };
+
+type Settings = {
+    pushGateway?: string;
+    intervalMs: number;
+    offsetMs: number;
+};
+
 // if this file is the entrypoint of the nodejs process
 if (require.main === module) {
     if (!process.env.UHTTP_LM_CLIENT_ID) {
@@ -15,39 +24,76 @@ if (require.main === module) {
     if (!process.env.UHTTP_LM_ZERO_HOP) {
         throw new Error("Missing 'UHTTP_LM_ZERO_HOP' env var.");
     }
-    if (!process.env.UHTTP_LM_LOCATION) {
-        log.warn("'UHTTP_LM_LOCATION' not set, using 'unset'.");
+    if (!process.env.UHTTP_LM_DISCOVERY_PLATFORM) {
+        throw new Error("Missing 'UHTTP_LM_DISCOVERY_PLATFORM' env var.");
     }
-    let location = process.env.UHTTP_LM_LOCATION || '';
-    location.trim();
-    if (!location) {
-        location = 'unset';
+    if (!process.env.UHTTP_LM_INTERVAL_MS) {
+        throw new Error("Missing 'UHTTP_LM_INTERVAL_MS' env var.");
+    }
+    if (!process.env.UHTTP_LM_OFFSET_MS) {
+        throw new Error("Missing 'UHTTP_LM_OFFSET_MS' env var.");
     }
     if (!process.env.UHTTP_LM_PUSH_GATEWAY) {
         log.warn("'UHTTP_LM_PUSH_GATEWAY' not set, disabling metrics pushing");
     }
-    const pushGateway = process.env.UHTTP_LM_PUSH_GATEWAY;
 
+    const uClientId = process.env.UHTTP_LM_CLIENT_ID;
+    const rpcProvider = process.env.UHTTP_LM_RPC_PROVIDER;
     const forceZeroHop = parseBooleanEnv(process.env.UHTTP_LM_ZERO_HOP);
-    const hops = forceZeroHop ? 0 : 1;
-    const ops = {
-        uhttpClientId: process.env.UHTTP_LM_CLIENT_ID,
-        rpcProvider: process.env.UHTTP_LM_RPC_PROVIDER,
+    const discoveryPlatformEndpoint = process.env.UHTTP_LM_DISCOVERY_PLATFORM;
+    const intervalMs = parseInt(process.env.UHTTP_LM_INTERVAL_MS);
+    if (!intervalMs) {
+        throw new Error("failed to parse 'UHTTP_LM_INTERVAL_MS' as integer value");
+    }
+    const offsetMs = parseInt(process.env.UHTTP_LM_OFFSET_MS);
+    if (!offsetMs) {
+        throw new Error("failed to parse 'UHTTP_LM_OFFSET_MS' as integer value");
+    }
+    const pushGateway = process.env.UHTTP_LM_PUSH_GATEWAY;
+
+    const uHTTPsettings = {
+        uClientId,
+        discoveryPlatformEndpoint,
         forceZeroHop,
+        rpcProvider,
+    };
+    const settings = {
+        pushGateway,
+        intervalMs,
+        offsetMs,
     };
-    const logOps = {
-        rpcProvider: ops.rpcProvider,
+    const logOpts = {
+        uHTTPsettings,
+        settings,
     };
-    log.info('Latency Monitor[%s] started with %o', Version, logOps);
+    log.info('Latency Monitor[%s] started with %o', Version, logOpts);
+
+    start(uHTTPsettings, settings);
+}
+
+function start(uHTTPsettings: UHTTPsettings, settings: Settings) {
+    const uClient = runner.init(uHTTPsettings.uClientId, uHTTPsettings);
+
+    setTimeout(() => {
+        tick(uClient, uHTTPsettings, settings);
+        setInterval(() => {
+            tick(uClient, uHTTPsettings, settings);
+        }, settings.intervalMs);
+    }, settings.offsetMs);
+    log.info('Delaying first tick by %dms offset', settings.offsetMs);
+}
 
+function tick(uClient: Routing.Client, uHTTPsettings: UHTTPsettings, settings: Settings) {
+    log.info('Executing latency tick - scheduled to execute every %dms', settings.intervalMs);
+    const hops = uHTTPsettings.forceZeroHop ? 0 : 1;
     runner
-        .once(ops)
-        .then(collectMetrics(hops, location))
-        .catch(reportError(hops, location))
-        .finally(pushMetrics(pushGateway));
+        .once(uClient, uHTTPsettings.rpcProvider)
+        .then(collectMetrics(hops))
+        .catch(reportError(hops))
+        .finally(pushMetrics(settings.pushGateway));
 }
 
-function collectMetrics(hops: number, location: string) {
+function collectMetrics(hops: number) {
     return function (metrics: runner.Durations) {
         const fetchSum = new prom.Summary({
             name: `uhttp_latency_milliseconds`,
@@ -79,23 +125,23 @@ function collectMetrics(hops: number, location: string) {
             labelNames: ['hops', 'location'] as const,
             percentiles: [0.5, 0.7, 0.9, 0.99],
         });
-        fetchSum.observe({ hops, location }, metrics.fetchDur);
-        rpcSum.observe({ hops, location }, metrics.rpcDur);
-        exitAppSum.observe({ hops, location }, metrics.exitAppDur);
-        segSum.observe({ hops, location }, metrics.segDur);
-        hoprSum.observe({ hops, location }, metrics.hoprDur);
+        fetchSum.observe({ hops }, metrics.fetchDur);
+        rpcSum.observe({ hops }, metrics.rpcDur);
+        exitAppSum.observe({ hops }, metrics.exitAppDur);
+        segSum.observe({ hops }, metrics.segDur);
+        hoprSum.observe({ hops }, metrics.hoprDur);
     };
 }
 
-function reportError(hops: number, location: string) {
+function reportError(hops: number) {
     return function (err: Error) {
         log.error('Error trying to check latency: %s', err);
         const errorSum = new prom.Summary({
             name: `uhttp_error`,
             help: 'Latency measure not possible due to error',
             labelNames: ['hops', 'location'] as const,
         });
-        errorSum.observe({ hops, location }, 0);
+        errorSum.observe({ hops }, 0);
     };
 }
 

src/runner.ts

@@ -1,22 +1,16 @@
 import { Routing } from '@hoprnet/uhttp-lib';
 
-type Ops = {
-    rpcProvider: string;
-    uhttpClientId: string;
-    forceZeroHop: boolean;
-};
-
 const READY_TIMEOUT = 10_000; // 10 sec
 
 export type Durations = {
     fetchDur: number;
 } & Routing.LatencyStatistics;
 
-export async function once(ops: Ops): Promise<Durations> {
-    const uClient = new Routing.Client(ops.uhttpClientId, {
-        forceZeroHop: ops.forceZeroHop,
-        measureLatency: true,
-    });
+export function init(uClientId: string, settings: Routing.Settings): Routing.Client {
+    return new Routing.Client(uClientId, { ...settings, measureLatency: true });
+}
+
+export async function once(uClient: Routing.Client, rpcProvider: string): Promise<Durations> {
     const id = Math.floor(Math.random() * 100);
     const payload = {
         jsonrpc: '2.0',
@@ -36,7 +30,7 @@ export async function once(ops: Ops): Promise<Durations> {
     };
 
     const startedAt = performance.now();
-    const res = await uClient.fetch(ops.rpcProvider, {
+    const res = await uClient.fetch(rpcProvider, {
         method: 'POST',
         body: JSON.stringify(payload),
         headers,