feat(validator-tagger): reduce clickhouse query memory usage, display validator list, remove ethstore exporter, add lido dvt, add beaconscore to validator view, use 90d period instead of all time

Author: peterbitfly

Makefile

@@ -7,7 +7,7 @@ LDFLAGS="-X ${PACKAGE}/version.Version=${VERSION} -X ${PACKAGE}/version.BuildDat
 CGO_CFLAGS="-O -D__BLST_PORTABLE__"
 CGO_CFLAGS_ALLOW="-O -D__BLST_PORTABLE__"
 
-all: explorer stats frontend-data-updater eth1indexer blobindexer ethstore-exporter rewards-exporter node-jobs-processor signatures notification-sender notification-collector user-service misc validator-tagger
+all: explorer stats frontend-data-updater eth1indexer blobindexer rewards-exporter node-jobs-processor signatures notification-sender notification-collector user-service misc validator-tagger
 
 lint:
 	echo 
@@ -34,9 +34,6 @@ stats:
 frontend-data-updater:
 	CGO_CFLAGS=${CGO_CFLAGS} CGO_CFLAGS_ALLOW=${CGO_CFLAGS_ALLOW} go build --ldflags=${LDFLAGS} -o bin/frontend-data-updater cmd/frontend-data-updater/main.go
 
-ethstore-exporter:
-	CGO_CFLAGS=${CGO_CFLAGS} CGO_CFLAGS_ALLOW=${CGO_CFLAGS_ALLOW} go build --ldflags=${LDFLAGS} -o bin/ethstore-exporter cmd/ethstore-exporter/main.go
-
 rewards-exporter:
 	CGO_CFLAGS=${CGO_CFLAGS} CGO_CFLAGS_ALLOW=${CGO_CFLAGS_ALLOW} go build --ldflags=${LDFLAGS} -o bin/rewards-exporter cmd/rewards-exporter/main.go
 

cmd/ethstore-exporter/main.go

@@ -519,7 +519,10 @@ func main() {
 
 			router.HandleFunc("/ethClients", handlers.EthClientsServices).Methods("GET")
 			router.HandleFunc("/entities", handlers.Entities).Methods("GET")
+			router.HandleFunc("/entities/data", handlers.EntitiesData).Methods("GET")
 			router.HandleFunc("/entity/{entity}/{subEntity}", handlers.EntityDetail).Methods("GET")
+			router.HandleFunc("/entity/{entity}/{subEntity}/subentities/data", handlers.EntitySubEntitiesData).Methods("GET")
+			router.HandleFunc("/entity/{entity}/{subEntity}/validators/data", handlers.EntityValidatorsData).Methods("GET")
 			router.HandleFunc("/relays", handlers.Relays).Methods("GET")
 			router.HandleFunc("/pools/rocketpool", handlers.PoolsRocketpool).Methods("GET")
 			router.HandleFunc("/pools/rocketpool/data/minipools", handlers.PoolsRocketpoolDataMinipools).Methods("GET")

cmd/explorer/main.go

@@ -22,7 +22,7 @@ func main() {
 	configPath := flag.String("config", "", "Path to the config file, if empty string defaults will be used")
 	versionFlag := flag.Bool("version", false, "Show version and exit")
 	scheduleFlag := flag.Bool("schedule", false, "Start scheduler loop (daily 10:00 UTC and hourly precompute)")
-	runFlag := flag.String("run", "", "Comma-separated steps to run on demand (import,lido,lido_csm,rocketpool,withdrawal_tagging,deposit_tagging,populate_validator_names,precompute,all)")
+	runFlag := flag.String("run", "", "Comma-separated steps to run on demand (import,lido,lido_csm,lido_simple_dvt,rocketpool,withdrawal_tagging,deposit_tagging,populate_validator_names,precompute,all)")
 	flag.Parse()
 
 	if *versionFlag {

cmd/validator-tagger/main.go

@@ -31,30 +31,57 @@ func FetchClickhouseParquet[T any](ctx context.Context, sql string, yield func(T
 	scheme := "https"
 	url := fmt.Sprintf("%s://%s:%d/?database=%s&enable_http_compression=1", scheme, chCfg.Host, 8443, chCfg.Name)
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, io.NopCloser(strings.NewReader(sql)))
-	if err != nil {
-		return fmt.Errorf("new request: %w", err)
-	}
-	// Basic auth
-	req.SetBasicAuth(chCfg.Username, chCfg.Password)
-	// We'll expect raw Parquet bytes in the body
-	req.Header.Set("Content-Type", "text/plain; charset=UTF-8")
-
 	client := &http.Client{Timeout: time.Second * 120}
-	resp, err := client.Do(req)
-	if err != nil {
-		return fmt.Errorf("http do: %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		defer resp.Body.Close()
-		b, _ := io.ReadAll(io.LimitReader(resp.Body, 8192))
-		return fmt.Errorf("clickhouse http status %d: %s", resp.StatusCode, string(b))
+
+	var data []byte
+	var lastErr error
+	for attempt := 1; attempt <= 5; attempt++ {
+		req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, io.NopCloser(strings.NewReader(sql)))
+		if err != nil {
+			return fmt.Errorf("new request: %w", err)
+		}
+		// Basic auth
+		req.SetBasicAuth(chCfg.Username, chCfg.Password)
+		// We'll expect raw Parquet bytes in the body
+		req.Header.Set("Content-Type", "text/plain; charset=UTF-8")
+
+		resp, err := client.Do(req)
+		if err != nil {
+			lastErr = fmt.Errorf("http do (attempt %d/5): %w", attempt, err)
+			logger.WithField("attempt", attempt).Warnf("clickhouse parquet http request failed: %v", err)
+		} else if resp.StatusCode != http.StatusOK {
+			b, _ := io.ReadAll(io.LimitReader(resp.Body, 8192))
+			_ = resp.Body.Close()
+			lastErr = fmt.Errorf("clickhouse http status %d (attempt %d/5): %s", resp.StatusCode, attempt, string(b))
+			logger.WithFields(map[string]interface{}{"attempt": attempt, "status": resp.StatusCode}).Warnf("clickhouse parquet http non-200: %s", strings.TrimSpace(string(b)))
+		} else {
+			logger.Info("fetched parquet http response")
+			// Read body fully
+			bodyBytes, err := io.ReadAll(resp.Body)
+			resp.Body.Close()
+			if err != nil {
+				lastErr = fmt.Errorf("read parquet http body (attempt %d/5): %w", attempt, err)
+				logger.WithField("attempt", attempt).Warnf("failed reading parquet http body: %v", err)
+			} else {
+				data = bodyBytes
+				break
+			}
+		}
+
+		if attempt < 5 {
+			// If context is cancelled, abort early
+			select {
+			case <-ctx.Done():
+				return fmt.Errorf("context canceled during clickhouse fetch: %w", ctx.Err())
+			case <-time.After(1 * time.Second):
+			}
+		}
 	}
-	logger.Info("fetched parquet http response")
-	defer resp.Body.Close()
-	data, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return fmt.Errorf("read parquet http body: %w", err)
+	if data == nil {
+		if lastErr != nil {
+			return lastErr
+		}
+		return fmt.Errorf("failed to fetch parquet data from clickhouse after 5 attempts")
 	}
 	// print the body size in megabytes
 	logger.Infof("fetched %d Mb of parquet data", len(data)/1024/1024)

db/clickhouse.go

@@ -103,7 +103,7 @@ func GetEntitiesTreemapData(period string) ([]types.EntityTreemapItem, error) {
 			logger.WithFields(map[string]interface{}{"period": period, "rows": len(cached)}).Debug("treemap: cache hit")
 			return cached, nil
 		}
-		logger.WithField("period", period).Debug("treemap: cache miss")
+		logger.WithField("period", period).Warn("treemap: cache miss")
 	}
 
 	// Fetch from DB (singleflight) and optionally warm cache
@@ -296,3 +296,26 @@ func GetSubEntitiesForEntities(entityNames []string, period string) ([]SubEntity
 	`, pq.Array(entityNames), period)
 	return subEntities, err
 }
+
+// GetSubEntityCountsForEntities returns the number of real sub-entities (sub_entity <> '-') per entity for the given period.
+func GetSubEntityCountsForEntities(entityNames []string, period string) (map[string]int, error) {
+	type countRow struct {
+		Entity string `db:"entity"`
+		Count  int    `db:"cnt"`
+	}
+	var rows []countRow
+	err := ReaderDb.Select(&rows, `
+		SELECT entity, COUNT(*) AS cnt
+		FROM validator_entities_data_periods
+		WHERE period = $2 AND entity = ANY($1) AND sub_entity <> '-'
+		GROUP BY entity
+	`, pq.Array(entityNames), period)
+	if err != nil {
+		return nil, err
+	}
+	result := make(map[string]int, len(rows))
+	for _, r := range rows {
+		result[r.Entity] = r.Count
+	}
+	return result, nil
+}