fix result codes update (#319)

* fix: add `observeRetrievalResultCodes` to dry-run Signed-off-by: Miroslav Bajtoš <[email protected]> * fix: day returned by InfluxDB query The function `observeRetrievalResultCodes()` includes the following statement in the InfluxDB query fetching `(day, code, rate)` data: |> aggregateWindow(every: 1d, fn: mean, createEmpty: false) Such query produces a list of values like this: ``` 2024-11-15T00:00:00Z,CONNECTION_REFUSED,0.0022313570194142725 2024-11-16T00:00:00Z,CONNECTION_REFUSED,0.002153071995819862 (...) 2025-02-12T00:00:00Z,CONNECTION_REFUSED,0.021266890041248942 2025-02-12T13:08:20.817239423Z,CONNECTION_REFUSED,0.02153170594662248 ``` Notice there are two rows for today (2025-02-12). One row contains data from yesterday (full day) and another row contain partial data from today. In this commit, I am fixing the query to correctly assign data points from yesterday to yesterday's date: |> aggregateWindow(every: 1d, fn: mean, createEmpty: false, timeSrc: "_start") The new query produces a list like this: ``` 2024-11-14T00:00:00Z,CONNECTION_REFUSED,0.0022313570194142725 2024-11-15T00:00:00Z,CONNECTION_REFUSED,0.002153071995819862 (...) 2025-02-11T00:00:00Z,CONNECTION_REFUSED,0.021266890041248942 2025-02-12T00:00:00Z,CONNECTION_REFUSED,0.02153170594662248 ``` This fixed the error introduced by cbb3bf1 (#316), where the SQL query fails with the following message: ``` ON CONFLICT DO UPDATE command cannot affect row a second time. Ensure that no rows proposed for insertion within the same command have duplicate constrained values. ``` See also InfluxDB documentation for `aggregateWindow()`: https://docs.influxdata.com/flux/v0/stdlib/universe/aggregatewindow/#timesrc Signed-off-by: Miroslav Bajtoš <[email protected]> * fixup! set INFLUXDB_TOKEN for dry-run Signed-off-by: Miroslav Bajtoš <[email protected]> --------- Signed-off-by: Miroslav Bajtoš <[email protected]> Co-authored-by: Srdjan <[email protected]>
CheckerNetwork · Feb 13, 2025 · 3c2c5b3 · 3c2c5b3
1 parent 79d59b2
commit 3c2c5b3
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 3 deletions.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -108,6 +108,7 @@ jobs:
       - run: npm run dry-run
         env:
             GLIF_TOKEN: ${{ secrets.GLIF_TOKEN }}
+            INFLUXDB_TOKEN: ${{ secrets.INFLUXDB_TOKEN }}
 
   docker-build:
     runs-on: ubuntu-latest

diff --git a/observer/bin/dry-run.js b/observer/bin/dry-run.js
@@ -1,10 +1,15 @@
 import * as SparkImpactEvaluator from '@filecoin-station/spark-impact-evaluator'
 import { ethers } from 'ethers'
+import assert from 'node:assert'
 
 import { RPC_URL, rpcHeaders } from '../lib/config.js'
-import { observeTransferEvents, observeScheduledRewards } from '../lib/observer.js'
+import { observeTransferEvents, observeScheduledRewards, observeRetrievalResultCodes } from '../lib/observer.js'
+import { createInflux } from '../lib/telemetry.js'
 import { getPgPools } from '@filecoin-station/spark-stats-db'
 
+const { INFLUXDB_TOKEN } = process.env
+assert(INFLUXDB_TOKEN, 'INFLUXDB_TOKEN required')
+
 const pgPools = await getPgPools()
 
 const fetchRequest = new ethers.FetchRequest(RPC_URL)
@@ -13,11 +18,15 @@ const provider = new ethers.JsonRpcProvider(fetchRequest, null, { polling: true
 
 const ieContract = new ethers.Contract(SparkImpactEvaluator.ADDRESS, SparkImpactEvaluator.ABI, provider)
 
+const { influx } = createInflux(INFLUXDB_TOKEN)
+const influxQueryApi = influx.getQueryApi('Filecoin Station')
+
 await pgPools.stats.query('DELETE FROM daily_reward_transfers')
 
 await Promise.all([
   observeTransferEvents(pgPools.stats, ieContract, provider),
-  observeScheduledRewards(pgPools, ieContract)
+  observeScheduledRewards(pgPools, ieContract),
+  observeRetrievalResultCodes(pgPools.stats, influxQueryApi)
 ])
 
 await pgPools.stats.end()
diff --git a/observer/lib/observer.js b/observer/lib/observer.js
@@ -113,7 +113,7 @@ export const observeRetrievalResultCodes = async (pgPoolStats, influxQueryApi) =
       |> range(start: 0)
       |> filter(fn: (r) => r["_measurement"] == "retrieval_stats_honest")
       |> filter(fn: (r) => strings.hasPrefix(v: r._field, prefix: "result_rate_"))
-      |> aggregateWindow(every: 1d, fn: mean, createEmpty: false)
+      |> aggregateWindow(every: 1d, fn: mean, createEmpty: false, timeSrc: "_start")
       |> keep(columns: ["_value", "_time", "_field"])
       |> map(fn: (r) => ({ r with _field: strings.replace(v: r._field, t: "result_rate_", u: "", i: 1) }))
   `)