refactor: move deposits-aggregation from exporter to statistics and r…

cmd/statistics/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"flag"
 	"fmt"
+	"log"
 	"math/big"
 	"strconv"
 	"strings"
@@ -23,13 +24,15 @@ import (
 )
 
 type options struct {
-	configPath                string
-	statisticsDayToExport     int64
-	statisticsDaysToExport    string
-	statisticsValidatorToggle bool
-	statisticsChartToggle     bool
-	statisticsGraffitiToggle  bool
-	resetStatus               bool
+	configPath                 string
+	statisticsDayToExport      int64
+	statisticsDaysToExport     string
+	statisticsValidatorToggle  bool
+	statisticsChartToggle      bool
+	statisticsGraffitiToggle   bool
+	statisticsDepositsToggle   bool
+	statisticsDepositsInterval time.Duration
+	resetStatus                bool
 }
 
 var opt = &options{}
@@ -41,6 +44,8 @@ func main() {
 	flag.BoolVar(&opt.statisticsValidatorToggle, "validators.enabled", false, "Toggle exporting validator statistics")
 	flag.BoolVar(&opt.statisticsChartToggle, "charts.enabled", false, "Toggle exporting chart series")
 	flag.BoolVar(&opt.statisticsGraffitiToggle, "graffiti.enabled", false, "Toggle exporting graffiti statistics")
+	flag.BoolVar(&opt.statisticsDepositsToggle, "deposits.enabled", false, "Toggle aggregating deposits")
+	flag.DurationVar(&opt.statisticsDepositsInterval, "deposits.interval", time.Hour*24, "Duration to wait between deposit aggregation")
 	flag.BoolVar(&opt.resetStatus, "validators.reset", false, "Export stats independet if they have already been exported previously")
 
 	versionFlag := flag.Bool("version", false, "Show version and exit")
@@ -241,6 +246,10 @@ func main() {
 
 	go statisticsLoop(rpcClient)
 
+	if opt.statisticsDepositsToggle {
+		go depositsLoop()
+	}
+
 	utils.WaitForCtrlC()
 
 	logrus.Println("exiting...")
@@ -281,6 +290,7 @@ func statisticsLoop(client rpc.Client) {
 			if lastExportedDayValidator != 0 {
 				lastExportedDayValidator++
 			}
+
 			if lastExportedDayValidator <= previousDay || lastExportedDayValidator == 0 {
 				for day := lastExportedDayValidator; day <= previousDay; day++ {
 					err := db.WriteValidatorStatisticsForDay(day, client)
@@ -353,6 +363,25 @@ func statisticsLoop(client rpc.Client) {
 	}
 }
 
+func depositsLoop() {
+	if opt.statisticsDepositsInterval < time.Minute {
+		log.Fatal(nil, "deposits.interval must be at least 1 minute", 0)
+	}
+	time.Sleep(time.Minute) // wait in case the process is in crashloop
+	for {
+		start := time.Now()
+		err := db.AggregateDeposits()
+		if err != nil {
+			logrus.Errorf("error aggregating deposits: %v", err)
+			services.ReportStatus("deposits_aggregator", err.Error(), nil)
+		} else {
+			logrus.WithFields(logrus.Fields{"duration": time.Since(start)}).Infof("aggregated deposits")
+			services.ReportStatus("deposits_aggregator", "Running", nil)
+		}
+		time.Sleep(opt.statisticsDepositsInterval)
+	}
+}
+
 func clearStatsStatusTable(day uint64) {
 	logrus.Infof("deleting validator_stats_status for day %v", day)
 	_, err := db.WriterDb.Exec("DELETE FROM validator_stats_status WHERE day = $1", day)

db/statistics.go

@@ -1878,3 +1878,47 @@ func CheckIfDayIsFinalized(day uint64) error {
 
 	return nil
 }
+
+func AggregateDeposits() error {
+	start := time.Now()
+	defer func() {
+		metrics.TaskDuration.WithLabelValues("statistics_aggregate_eth1_deposits").Observe(time.Since(start).Seconds())
+	}()
+	_, err := WriterDb.Exec(`
+		INSERT INTO eth1_deposits_aggregated (from_address, amount, validcount, invalidcount, slashedcount, totalcount, activecount, pendingcount, voluntary_exit_count)
+		SELECT
+			eth1.from_address,
+			SUM(eth1.amount) as amount,
+			SUM(eth1.validcount) AS validcount,
+			SUM(eth1.invalidcount) AS invalidcount,
+			COUNT(CASE WHEN v.status = 'slashed' THEN 1 END) AS slashedcount,
+			COUNT(v.pubkey) AS totalcount,
+			COUNT(CASE WHEN v.status = 'active_online' OR v.status = 'active_offline' THEN 1 END) as activecount,
+			COUNT(CASE WHEN v.status = 'deposited' THEN 1 END) AS pendingcount,
+			COUNT(CASE WHEN v.status = 'exited' THEN 1 END) AS voluntary_exit_count
+		FROM (
+			SELECT
+				from_address,
+				publickey,
+				SUM(amount) AS amount,
+				COUNT(CASE WHEN valid_signature = 't' THEN 1 END) AS validcount,
+				COUNT(CASE WHEN valid_signature = 'f' THEN 1 END) AS invalidcount
+			FROM eth1_deposits
+			GROUP BY from_address, publickey
+		) eth1
+		LEFT JOIN (SELECT pubkey, status FROM validators) v ON v.pubkey = eth1.publickey
+		GROUP BY eth1.from_address
+		ON CONFLICT (from_address) DO UPDATE SET
+			amount               = excluded.amount,
+			validcount           = excluded.validcount,
+			invalidcount         = excluded.invalidcount,
+			slashedcount         = excluded.slashedcount,
+			totalcount           = excluded.totalcount,
+			activecount          = excluded.activecount,
+			pendingcount         = excluded.pendingcount,
+			voluntary_exit_count = excluded.voluntary_exit_count`)
+	if err != nil && err != sql.ErrNoRows {
+		return err
+	}
+	return nil
+}

exporter/eth1.go

@@ -2,14 +2,12 @@ package exporter
 
 import (
 	"context"
-	"database/sql"
 	"fmt"
 	"math/big"
 	"regexp"
 	"time"
 
 	"github.com/gobitfly/eth2-beaconchain-explorer/db"
-	"github.com/gobitfly/eth2-beaconchain-explorer/metrics"
 	"github.com/gobitfly/eth2-beaconchain-explorer/types"
 	"github.com/gobitfly/eth2-beaconchain-explorer/utils"
 
@@ -128,15 +126,6 @@ func eth1DepositsExporter() {
 			continue
 		}
 
-		if len(depositsToSave) > 0 {
-			err = aggregateDeposits()
-			if err != nil {
-				logger.WithError(err).Errorf("error saving eth1-deposits-leaderboard")
-				time.Sleep(time.Second * 5)
-				continue
-			}
-		}
-
 		// make sure we are progressing even if there are no deposits in the last batch
 		lastFetchedBlock = toBlock
 
@@ -373,47 +362,3 @@ func eth1BatchRequestHeadersAndTxs(blocksToFetch []uint64, txsToFetch []string)
 
 	return headers, txs, nil
 }
-
-func aggregateDeposits() error {
-	start := time.Now()
-	defer func() {
-		metrics.TaskDuration.WithLabelValues("exporter_aggregate_eth1_deposits").Observe(time.Since(start).Seconds())
-	}()
-	_, err := db.WriterDb.Exec(`
-		INSERT INTO eth1_deposits_aggregated (from_address, amount, validcount, invalidcount, slashedcount, totalcount, activecount, pendingcount, voluntary_exit_count)
-		SELECT
-			eth1.from_address,
-			SUM(eth1.amount) as amount,
-			SUM(eth1.validcount) AS validcount,
-			SUM(eth1.invalidcount) AS invalidcount,
-			COUNT(CASE WHEN v.status = 'slashed' THEN 1 END) AS slashedcount,
-			COUNT(v.pubkey) AS totalcount,
-			COUNT(CASE WHEN v.status = 'active_online' OR v.status = 'active_offline' THEN 1 END) as activecount,
-			COUNT(CASE WHEN v.status = 'deposited' THEN 1 END) AS pendingcount,
-			COUNT(CASE WHEN v.status = 'exited' THEN 1 END) AS voluntary_exit_count
-		FROM (
-			SELECT 
-				from_address,
-				publickey,
-				SUM(amount) AS amount,
-				COUNT(CASE WHEN valid_signature = 't' THEN 1 END) AS validcount,
-				COUNT(CASE WHEN valid_signature = 'f' THEN 1 END) AS invalidcount
-			FROM eth1_deposits
-			GROUP BY from_address, publickey
-		) eth1
-		LEFT JOIN (SELECT pubkey, status FROM validators) v ON v.pubkey = eth1.publickey
-		GROUP BY eth1.from_address
-		ON CONFLICT (from_address) DO UPDATE SET
-			amount               = excluded.amount,
-			validcount           = excluded.validcount,
-			invalidcount         = excluded.invalidcount,
-			slashedcount         = excluded.slashedcount,
-			totalcount           = excluded.totalcount,
-			activecount          = excluded.activecount,
-			pendingcount         = excluded.pendingcount,
-			voluntary_exit_count = excluded.voluntary_exit_count`)
-	if err != nil && err != sql.ErrNoRows {
-		return nil
-	}
-	return err
-}

templates/eth1DepositsLeaderboard.html

@@ -195,6 +195,7 @@
         <div class="d-md-flex py-2 justify-content-md-between">
           <div class="heading">
             <h1 class="h4 mb-1 mb-md-0"><i class="fas fa-file-signature mr-2"></i>Validator Deposit Leaderboard</h1>
+            <div>Data on this page only gets updated once a day.</div>
           </div>
           <nav aria-label="breadcrumb">
             <ol class="breadcrumb font-size-1 mb-0" style="padding:0; background-color:transparent;">