(BIDS-2261) improve statistics export performance by gathering data i…

…n parallel and using postgres copy from for bulk data loading

cmd/eth1indexer/main.go

@@ -25,7 +25,7 @@ import (
 
 	"github.com/coocood/freecache"
 	"github.com/ethereum/go-ethereum/common"
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/shopspring/decimal"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/sync/errgroup"

cmd/ethstore-exporter/main.go

@@ -11,7 +11,7 @@ import (
 	"strconv"
 	"strings"
 
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/sirupsen/logrus"
 )
 

cmd/explorer/main.go

@@ -34,7 +34,7 @@ import (
 
 	"github.com/gorilla/csrf"
 	"github.com/gorilla/mux"
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/phyber/negroni-gzip/gzip"
 	"github.com/stripe/stripe-go/v72"
 	"github.com/urfave/negroni"

cmd/frontend-data-updater/main.go

@@ -12,7 +12,7 @@ import (
 	"fmt"
 	"math/big"
 
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/sirupsen/logrus"
 
 	"net/http"

cmd/misc/main.go

@@ -17,7 +17,7 @@ import (
 	"time"
 
 	"github.com/coocood/freecache"
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	utilMath "github.com/protolambda/zrnt/eth2/util/math"
 	"golang.org/x/sync/errgroup"
 

cmd/node-jobs-processor/main.go

@@ -10,7 +10,7 @@ import (
 	"fmt"
 	"time"
 
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 
 	"github.com/sirupsen/logrus"
 )

cmd/notification-collector/main.go

@@ -19,7 +19,7 @@ import (
 	_ "eth2-exporter/docs"
 	_ "net/http/pprof"
 
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 )
 
 func main() {

cmd/notification-sender/main.go

@@ -19,7 +19,7 @@ import (
 	_ "eth2-exporter/docs"
 	_ "net/http/pprof"
 
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 )
 
 func main() {

cmd/rewards-exporter/main.go

@@ -14,7 +14,7 @@ import (
 
 	eth_rewards "github.com/gobitfly/eth-rewards"
 	"github.com/gobitfly/eth-rewards/beacon"
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/sirupsen/logrus"
 	"golang.org/x/sync/errgroup"
 )

cmd/signatures/main.go

@@ -14,7 +14,7 @@ import (
 	"net/http"
 	"time"
 
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/sirupsen/logrus"
 
 	_ "net/http/pprof"

cmd/statistics/main.go

@@ -14,21 +14,18 @@ import (
 	"strings"
 	"time"
 
-	_ "github.com/jackc/pgx/v4/stdlib"
+	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/sirupsen/logrus"
 )
 
 type options struct {
-	configPath                    string
-	statisticsDayToExport         int64
-	statisticsDaysToExport        string
-	statisticsValidatorToggle     bool
-	statisticsResetColumns        string
-	statisticsChartToggle         bool
-	statisticsGraffitiToggle      bool
-	concurrencyTotal              uint64
-	concurrencyCl                 uint64
-	concurrencyFailedAttestations uint64
+	configPath                string
+	statisticsDayToExport     int64
+	statisticsDaysToExport    string
+	statisticsValidatorToggle bool
+	statisticsChartToggle     bool
+	statisticsGraffitiToggle  bool
+	resetStatus               bool
 }
 
 var opt = &options{}
@@ -38,12 +35,9 @@ func main() {
 	flag.Int64Var(&opt.statisticsDayToExport, "statistics.day", -1, "Day to export statistics (will export the day independent if it has been already exported or not")
 	flag.StringVar(&opt.statisticsDaysToExport, "statistics.days", "", "Days to export statistics (will export the day independent if it has been already exported or not")
 	flag.BoolVar(&opt.statisticsValidatorToggle, "validators.enabled", false, "Toggle exporting validator statistics")
-	flag.StringVar(&opt.statisticsResetColumns, "validators.reset", "", "validator_stats_status columns to reset. Comma separated. Use 'all' for complete resync.")
 	flag.BoolVar(&opt.statisticsChartToggle, "charts.enabled", false, "Toggle exporting chart series")
 	flag.BoolVar(&opt.statisticsGraffitiToggle, "graffiti.enabled", false, "Toggle exporting graffiti statistics")
-	flag.Uint64Var(&opt.concurrencyTotal, "concurrency.total", 10, "Concurrency to use when writing total rewards/performance postgres queries")
-	flag.Uint64Var(&opt.concurrencyCl, "concurrency.cl", 50, "Concurrency to use when writing cl postgres queries")
-	flag.Uint64Var(&opt.concurrencyFailedAttestations, "concurrency.fa", 10, "Concurrency to use when fetching failed attestaations from bt")
+	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")
 	flag.Parse()
@@ -142,9 +136,11 @@ func main() {
 			logrus.Infof("exporting validator statistics for days %v-%v", firstDay, lastDay)
 			for d := firstDay; d <= lastDay; d++ {
 
-				clearStatsStatusTable(d, opt.statisticsResetColumns)
+				if opt.resetStatus {
+					clearStatsStatusTable(d)
+				}
 
-				err = db.WriteValidatorStatisticsForDay(uint64(d), opt.concurrencyTotal, opt.concurrencyCl, opt.concurrencyFailedAttestations)
+				err = db.WriteValidatorStatisticsForDay(uint64(d))
 				if err != nil {
 					utils.LogError(err, fmt.Errorf("error exporting stats for day %v", d), 0)
 					break
@@ -182,9 +178,11 @@ func main() {
 	} else if opt.statisticsDayToExport >= 0 {
 
 		if opt.statisticsValidatorToggle {
-			clearStatsStatusTable(uint64(opt.statisticsDayToExport), opt.statisticsResetColumns)
+			if opt.resetStatus {
+				clearStatsStatusTable(uint64(opt.statisticsDayToExport))
+			}
 
-			err = db.WriteValidatorStatisticsForDay(uint64(opt.statisticsDayToExport), opt.concurrencyTotal, opt.concurrencyCl, opt.concurrencyFailedAttestations)
+			err = db.WriteValidatorStatisticsForDay(uint64(opt.statisticsDayToExport))
 			if err != nil {
 				utils.LogError(err, fmt.Errorf("error exporting stats for day %v", opt.statisticsDayToExport), 0)
 			}
@@ -211,14 +209,14 @@ func main() {
 		return
 	}
 
-	go statisticsLoop(opt.concurrencyTotal, opt.concurrencyCl, opt.concurrencyFailedAttestations)
+	go statisticsLoop()
 
 	utils.WaitForCtrlC()
 
 	logrus.Println("exiting...")
 }
 
-func statisticsLoop(concurrencyTotal uint64, concurrencyCl uint64, concurrencyFailedAttestations uint64) {
+func statisticsLoop() {
 	for {
 
 		latestEpoch := services.LatestFinalizedEpoch()
@@ -254,7 +252,7 @@ func statisticsLoop(concurrencyTotal uint64, concurrencyCl uint64, concurrencyFa
 			}
 			if lastExportedDayValidator <= previousDay || lastExportedDayValidator == 0 {
 				for day := lastExportedDayValidator; day <= previousDay; day++ {
-					err := db.WriteValidatorStatisticsForDay(day, concurrencyTotal, concurrencyCl, concurrencyFailedAttestations)
+					err := db.WriteValidatorStatisticsForDay(day)
 					if err != nil {
 						utils.LogError(err, fmt.Errorf("error exporting stats for day %v", day), 0)
 						break
@@ -305,23 +303,10 @@ func statisticsLoop(concurrencyTotal uint64, concurrencyCl uint64, concurrencyFa
 	}
 }
 
-func clearStatsStatusTable(day uint64, columns string) {
-	if columns == "all" {
-		logrus.Infof("Delete validator_stats_status for day %v", day)
-		_, err := db.WriterDb.Exec("DELETE FROM validator_stats_status WHERE day = $1", day)
-		if err != nil {
-			logrus.Fatalf("error resetting status for day %v: %v", day, err)
-		}
-	} else if len(columns) > 0 {
-		logrus.Infof("Resetting columns %v of validator_stats_status for day %v ", columns, day)
-		cols := strings.Join(strings.Split(columns, ","), " = false,")
-		_, err := db.WriterDb.Exec(fmt.Sprintf(`
-			UPDATE validator_stats_status
-			SET %v = false
-			WHERE day = $1
-		`, cols), day)
-		if err != nil {
-			logrus.Fatalf("error resetting status for day %v: %v", day, err)
-		}
+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)
+	if err != nil {
+		logrus.Fatalf("error resetting status for day %v: %v", day, err)
 	}
 }

db/bigtable.go

@@ -1485,7 +1485,7 @@ func (bigtable *Bigtable) GetValidatorMissedAttestationsCount(validators []uint6
 		return nil, err
 	}
 
-	logger.Infof("retrieved missed attestation history for epochs %v - %v", firstEpoch, lastEpoch)
+	// logger.Infof("retrieved missed attestation history for epochs %v - %v", firstEpoch, lastEpoch)
 
 	for validator, attestations := range data {
 		if len(attestations) == 0 {
@@ -1629,7 +1629,7 @@ func (bigtable *Bigtable) GetValidatorBalanceStatistics(validators []uint64, sta
 		}
 		vals := validators[i:upperBound]
 
-		logrus.Infof("retrieving validator balance stats for validators %v - %v", vals[0], vals[len(vals)-1])
+		// logrus.Infof("retrieving validator balance stats for validators %v - %v", vals[0], vals[len(vals)-1])
 
 		res, err := bigtable.GetValidatorBalanceHistory(vals, startEpoch, endEpoch)
 		if err != nil {

db/db.go

@@ -30,7 +30,7 @@ import (
 
 	"eth2-exporter/rpc"
 
-	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/jackc/pgx/v5/pgxpool"
 )
 
 //go:embed migrations/*.sql