forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-28 07:36:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Implementation of entries and storage metrics

Browse source
This commit is contained in:
Rafael Sampaio 2024-10-26 20:44:03 -03:00 committed by Chen Kai
parent bf673d4695
commit c8f6feb7ae
6 changed files with 296 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
cmd/shisui/main.go
View file

@ -127,6 +127,7 @@ func shisui(ctx *cli.Context) error {
// Start system runtime metrics collection
go metrics.CollectProcessMetrics(3 * time.Second)
go metrics.CollectPortalMetrics(5*time.Second, ctx.StringSlice(utils.PortalNetworksFlag.Name), ctx.String(utils.PortalDataDirFlag.Name))
if metrics.Enabled {
storageCapacity = metrics.NewRegisteredGauge("portal/storage_capacity", nil)
@ -456,7 +457,7 @@ func initState(config Config, server *rpc.Server, conn discover.UDPConn, localNo
if err != nil {
return nil, err
}
stateStore := state.NewStateStorage(contentStorage)
stateStore := state.NewStateStorage(contentStorage, db)
contentQueue := make(chan *discover.ContentElement, 50)
protocol, err := discover.NewPortalProtocol(config.Protocol, portalwire.State, config.PrivateKey, conn, localNode, discV5, stateStore, contentQueue)

82
metrics/portal_metrics.go Normal file
View file

@ -0,0 +1,82 @@
package metrics
import (
"os"
"path"
"slices"
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover/portalwire"
)
type networkFileMetric struct {
filename string
metric Gauge
file *os.File
network string
}
// CollectPortalMetrics periodically collects various metrics about system entities.
func CollectPortalMetrics(refresh time.Duration, networks []string, dataDir string) {
// Short circuit if the metrics system is disabled
if !Enabled {
return
}
// Define the various metrics to collect
var (
historyTotalStorage = GetOrRegisterGauge("portal/history/total_storage", nil)
beaconTotalStorage = GetOrRegisterGauge("portal/beacon/total_storage", nil)
stateTotalStorage = GetOrRegisterGauge("portal/state/total_storage", nil)
)
var metricsArr []*networkFileMetric
if slices.Contains(networks, portalwire.History.Name()) {
dbPath := path.Join(dataDir, portalwire.History.Name())
metricsArr = append(metricsArr, &networkFileMetric{
filename: path.Join(dbPath, portalwire.History.Name()+".sqlite"),
metric: historyTotalStorage,
network: portalwire.History.Name(),
})
}
if slices.Contains(networks, portalwire.Beacon.Name()) {
dbPath := path.Join(dataDir, portalwire.Beacon.Name())
metricsArr = append(metricsArr, &networkFileMetric{
filename: path.Join(dbPath, portalwire.Beacon.Name()+".sqlite"),
metric: beaconTotalStorage,
network: portalwire.Beacon.Name(),
})
}
if slices.Contains(networks, portalwire.State.Name()) {
dbPath := path.Join(dataDir, portalwire.State.Name())
metricsArr = append(metricsArr, &networkFileMetric{
filename: path.Join(dbPath, portalwire.State.Name()+".sqlite"),
metric: stateTotalStorage,
network: portalwire.State.Name(),
})
}
for {
for _, m := range metricsArr {
var err error = nil
if m.file == nil {
m.file, err = os.OpenFile(m.filename, os.O_RDONLY, 0600)
if err != nil {
log.Debug("Could not open file", "network", m.network, "file", m.filename, "metric", "total_storage", "err", err)
}
}
if m.file != nil && err == nil {
stat, err := m.file.Stat()
if err != nil {
log.Warn("Could not get file stat", "network", m.network, "file", m.filename, "metric", "total_storage", "err", err)
}
if err == nil {
m.metric.Update(stat.Size())
}
}
}
time.Sleep(refresh)
}
}

58
portalnetwork/beacon/storage.go
View file

@ -6,6 +6,7 @@ import (
"database/sql"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/holiman/uint256"
"github.com/protolambda/zrnt/eth2/beacon/common"
"github.com/protolambda/ztyp/codec"
@ -15,6 +16,17 @@ import (
const BytesInMB uint64 = 1000 * 1000
var (
radiusRatio metrics.GaugeFloat64
entriesCount metrics.Gauge
contentStorageUsage metrics.Gauge
)
const (
countEntrySql = "SELECT COUNT(1) FROM beacon;"
contentStorageUsageSql = "SELECT SUM( length(content_value) ) FROM beacon;"
)
type BeaconStorage struct {
storageCapacityInBytes uint64
db *sql.DB
@ -41,6 +53,44 @@ func NewBeaconStorage(config storage.PortalStorageConfig) (storage.ContentStorag
if err := bs.setup(); err != nil {
return nil, err
}
if metrics.Enabled {
radiusRatio = metrics.NewRegisteredGaugeFloat64("portal/beacon/radius_ratio", nil)
radiusRatio.Update(1)
entriesCount = metrics.NewRegisteredGauge("portal/beacon/entry_count", nil)
log.Info("Counting entities in beacon storage for metrics")
count, err := config.DB.Prepare(countEntrySql)
if err != nil {
log.Error("Querry preparation error", "network", config.NetworkName, "metric", "entry_count", "err", err)
return nil, err
}
var res *int64 = new(int64)
q := count.QueryRow()
if q.Err() != nil {
log.Error("Querry execution error", "network", config.NetworkName, "metric", "entry_count", "err", err)
return nil, err
} else {
q.Scan(res)
}
entriesCount.Update(*res)
contentStorageUsage = metrics.NewRegisteredGauge("portal/beacon/content_storage", nil)
log.Info("Counting storage usage (bytes) in beacon for metrics")
str, err := config.DB.Prepare(contentStorageUsageSql)
if err != nil {
log.Error("Querry preparation error", "network", config.NetworkName, "metric", "content_storage", "err", err)
return nil, err
}
var resStr *int64 = new(int64)
q = str.QueryRow()
if q.Err() != nil {
log.Error("Querry execution error", "network", config.NetworkName, "metric", "content_storage", "err", err)
return nil, err
} else {
q.Scan(resStr)
}
contentStorageUsage.Update(int64(*resStr))
}
return bs, nil
}
@ -175,10 +225,18 @@ func (bs *BeaconStorage) getLcUpdateValueByRange(start, end uint64) ([]byte, err
func (bs *BeaconStorage) putContentValue(contentId, contentKey, value []byte) error {
length := 32 + len(contentKey) + len(value)
_, err := bs.db.ExecContext(context.Background(), InsertQueryBeacon, contentId, contentKey, value, length)
if metrics.Enabled && err != nil {
entriesCount.Inc(1)
contentStorageUsage.Inc(int64(len(value)))
}
return err
}
func (bs *BeaconStorage) putLcUpdate(period uint64, value []byte) error {
_, err := bs.db.ExecContext(context.Background(), InsertLCUpdateQuery, period, value, 0, len(value))
if metrics.Enabled && err != nil {
entriesCount.Inc(1)
contentStorageUsage.Inc(int64(len(value)))
}
return err
}

94
portalnetwork/history/storage.go
View file

@ -22,7 +22,9 @@ import (
)
var (
radiusRatio metrics.GaugeFloat64
radiusRatio metrics.GaugeFloat64
entriesCount metrics.Gauge
contentStorageUsage metrics.Gauge
)
const (
@ -112,10 +114,7 @@ func NewHistoryStorage(config storage.PortalStorageConfig) (storage.ContentStora
log: log.New("storage", config.NetworkName),
}
hs.radius.Store(storage.MaxDistance)
if metrics.Enabled {
radiusRatio = metrics.NewRegisteredGaugeFloat64("portal/radius_ratio", nil)
radiusRatio.Update(1)
}
err := hs.createTable()
if err != nil {
return nil, err
@ -125,6 +124,28 @@ func NewHistoryStorage(config storage.PortalStorageConfig) (storage.ContentStora
// Check whether we already have data, and use it to set radius
// necessary to test NetworkName==history because state also initialize HistoryStorage
if metrics.Enabled && strings.ToLower(config.NetworkName) == "history" {
radiusRatio = metrics.NewRegisteredGaugeFloat64("portal/history/radius_ratio", nil)
radiusRatio.Update(1)
entriesCount = metrics.NewRegisteredGauge("portal/history/entry_count", nil)
log.Info("Counting entities in history storage for metrics")
count, err := hs.ContentCount()
if err != nil {
log.Debug("Querry execution error", "network", config.NetworkName, "metric", "entry_count", "err", err)
}
entriesCount.Update(int64(count))
contentStorageUsage = metrics.NewRegisteredGauge("portal/history/content_storage", nil)
log.Info("Counting storage usage (bytes) in history for metrics")
str, err := hs.ContentSize()
if err != nil {
log.Debug("Querry execution error", "network", config.NetworkName, "metric", "content_storage", "err", err)
}
contentStorageUsage.Update(int64(str))
}
return hs, err
}
@ -195,6 +216,10 @@ func (p *ContentStorage) put(contentId []byte, content []byte) PutResult {
return PutResult{pruned: true, count: count}
}
if metrics.Enabled {
entriesCount.Inc(1)
contentStorageUsage.Inc(int64(len(content)))
}
return PutResult{}
}
@ -289,6 +314,23 @@ func (p *ContentStorage) ContentSize() (uint64, error) {
return p.queryRowUint64(sql)
}
func (p *ContentStorage) SizeByKey(contentId []byte) (uint64, error) {
sql := "SELECT SUM( length(value) ) FROM kvstore WHERE key = " + string(contentId) + ";"
return p.queryRowUint64(sql)
}
func (p *ContentStorage) SizeByKeys(ids [][]byte) (uint64, error) {
sql := "SELECT SUM( length(value) ) FROM kvstore WHERE key IN (?" + strings.Repeat(", ?", len(ids)-1) + ");"
return p.queryRowUint64(sql)
}
func (p *ContentStorage) SizeOutRadius(radius *uint256.Int) (uint64, error) {
sql := "SELECT SUM( length(value) ) FROM kvstore WHERE greater(xor(key, (?1)), (?2)) = 1;"
var size uint64
err := p.sqliteDB.QueryRow(sql, p.nodeId[:], radius.Bytes()).Scan(&size)
return size, err
}
func (p *ContentStorage) queryRowUint64(sqlStr string) (uint64, error) {
// sql := "SELECT SUM(length(value)) FROM kvstore"
stmt, err := p.sqliteDB.Prepare(sqlStr)
@ -423,11 +465,31 @@ func (p *ContentStorage) deleteContentFraction(fraction float64) (deleteCount in
}
func (p *ContentStorage) del(contentId []byte) error {
_, err := p.delStmt.Exec(contentId)
var sizeDel uint64
var err error
if metrics.Enabled {
sizeDel, err = p.SizeByKey(contentId)
if err != nil {
return err
}
}
_, err = p.delStmt.Exec(contentId)
if metrics.Enabled && err != nil {
entriesCount.Dec(1)
contentStorageUsage.Dec(int64(sizeDel))
}
return err
}
func (p *ContentStorage) batchDel(ids [][]byte) error {
var sizeDel uint64
var err error
if metrics.Enabled {
sizeDel, err = p.SizeByKeys(ids)
if err != nil {
return err
}
}
query := "DELETE FROM kvstore WHERE key IN (?" + strings.Repeat(", ?", len(ids)-1) + ")"
args := make([]interface{}, len(ids))
for i, id := range ids {
@ -435,7 +497,11 @@ func (p *ContentStorage) batchDel(ids [][]byte) error {
}
// delete items
_, err := p.sqliteDB.Exec(query, args...)
_, err = p.sqliteDB.Exec(query, args...)
if metrics.Enabled && err != nil {
entriesCount.Dec(int64(len(args)))
contentStorageUsage.Dec(int64(sizeDel))
}
return err
}
@ -447,12 +513,24 @@ func (p *ContentStorage) ReclaimSpace() error {
}
func (p *ContentStorage) deleteContentOutOfRadius(radius *uint256.Int) error {
var sizeDel uint64
var err error
if metrics.Enabled {
sizeDel, err = p.SizeOutRadius(radius)
if err != nil {
return err
}
}
res, err := p.sqliteDB.Exec(deleteOutOfRadiusStmt, p.nodeId[:], radius.Bytes())
if err != nil {
return err
}
count, _ := res.RowsAffected()
count, err := res.RowsAffected()
p.log.Trace("delete items", "count", count)
if metrics.Enabled && err != nil {
entriesCount.Dec(count)
contentStorageUsage.Dec(int64(sizeDel))
}
return err
}

69
portalnetwork/state/storage.go
View file

@ -3,15 +3,28 @@ package state
import (
"bytes"
"crypto/sha256"
"database/sql"
"errors"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/portalnetwork/storage"
"github.com/holiman/uint256"
"github.com/protolambda/ztyp/codec"
)
var (
radiusRatio metrics.GaugeFloat64
entriesCount metrics.Gauge
contentStorageUsage metrics.Gauge
)
const (
countEntrySql = "SELECT COUNT(1) FROM kvstore;"
conentStorageUsageSql = "SELECT SUM( length(value) ) FROM kvstore;"
)
func defaultContentIdFunc(contentKey []byte) []byte {
digest := sha256.Sum256(contentKey)
return digest[:]
@ -21,14 +34,57 @@ var _ storage.ContentStorage = &StateStorage{}
type StateStorage struct {
store storage.ContentStorage
db *sql.DB
log log.Logger
}
func NewStateStorage(store storage.ContentStorage) *StateStorage {
return &StateStorage{
func NewStateStorage(store storage.ContentStorage, db *sql.DB) *StateStorage {
storage := &StateStorage{
store: store,
db: db,
log: log.New("storage", "state"),
}
if metrics.Enabled {
radiusRatio = metrics.NewRegisteredGaugeFloat64("portal/state/radius_ratio", nil)
radiusRatio.Update(1)
entriesCount = metrics.NewRegisteredGauge("portal/state/entry_count", nil)
log.Info("Counting entities in state storage for metrics")
count, err := db.Prepare(countEntrySql)
if err != nil {
log.Error("Querry preparation error", "network", "state", "metric", "entry_count", "err", err)
return nil
}
var res *int64 = new(int64)
q := count.QueryRow()
if q.Err() != nil {
log.Error("Querry execution error", "network", "state", "metric", "entry_count", "err", err)
return nil
} else {
q.Scan(&res)
}
entriesCount.Update(*res)
contentStorageUsage = metrics.NewRegisteredGauge("portal/state/content_storage", nil)
log.Info("Counting storage usage (bytes) in state for metrics")
str, err := db.Prepare(conentStorageUsageSql)
if err != nil {
log.Error("Querry preparation error", "network", "state", "metric", "content_storage", "err", err)
return nil
}
var resStr *int64 = new(int64)
q = str.QueryRow()
if q.Err() != nil {
log.Error("Querry execution error", "network", "state", "metric", "content_storage", "err", err)
return nil
} else {
q.Scan(resStr)
}
contentStorageUsage.Update(int64(*resStr))
}
return storage
}
// Get implements storage.ContentStorage.
@ -84,6 +140,9 @@ func (s *StateStorage) putAccountTrieNode(contentKey []byte, contentId []byte, c
err = s.store.Put(contentId, contentId, contentValueBuf.Bytes())
if err != nil {
s.log.Error("failed to save data after validate", "type", contentKey[0], "key", contentKey[1:], "value", content)
} else if metrics.Enabled {
entriesCount.Inc(1)
contentStorageUsage.Inc(int64(len(content)))
}
return nil
}
@ -118,6 +177,9 @@ func (s *StateStorage) putContractStorageTrieNode(contentKey []byte, contentId [
err = s.store.Put(contentId, contentId, contentValueBuf.Bytes())
if err != nil {
s.log.Error("failed to save data after validate", "type", contentKey[0], "key", contentKey[1:], "value", content)
} else if metrics.Enabled {
entriesCount.Inc(1)
contentStorageUsage.Inc(int64(len(content)))
}
return nil
}
@ -148,6 +210,9 @@ func (s *StateStorage) putContractBytecode(contentKey []byte, contentId []byte,
err = s.store.Put(contentId, contentId, contentValueBuf.Bytes())
if err != nil {
s.log.Error("failed to save data after validate", "type", contentKey[0], "key", contentKey[1:], "value", content)
} else if metrics.Enabled {
entriesCount.Inc(1)
contentStorageUsage.Inc(int64(len(content)))
}
return nil
}

2
portalnetwork/state/storage_test.go
View file

@ -10,7 +10,7 @@ import (
func TestStorage(t *testing.T) {
storage := storage.NewMockStorage()
stateStorage := NewStateStorage(storage)
stateStorage := NewStateStorage(storage, nil)
testfiles := []string{"account_trie_node.yaml", "contract_storage_trie_node.yaml", "contract_bytecode.yaml"}
for _, file := range testfiles {
cases, err := getTestCases(file)