From c8f6feb7aedd9166f1c9bbf66826cc6f45a566fa Mon Sep 17 00:00:00 2001 From: Rafael Sampaio <5679073+r4f4ss@users.noreply.github.com> Date: Sat, 26 Oct 2024 20:44:03 -0300 Subject: [PATCH] Implementation of entries and storage metrics --- cmd/shisui/main.go | 3 +- metrics/portal_metrics.go | 82 +++++++++++++++++++++++++ portalnetwork/beacon/storage.go | 58 ++++++++++++++++++ portalnetwork/history/storage.go | 94 ++++++++++++++++++++++++++--- portalnetwork/state/storage.go | 69 ++++++++++++++++++++- portalnetwork/state/storage_test.go | 2 +- 6 files changed, 296 insertions(+), 12 deletions(-) create mode 100644 metrics/portal_metrics.go diff --git a/cmd/shisui/main.go b/cmd/shisui/main.go index 6977180a4d..ea2d918e6d 100644 --- a/cmd/shisui/main.go +++ b/cmd/shisui/main.go @@ -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) diff --git a/metrics/portal_metrics.go b/metrics/portal_metrics.go new file mode 100644 index 0000000000..1d4cb3bcd9 --- /dev/null +++ b/metrics/portal_metrics.go @@ -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) + } +} diff --git a/portalnetwork/beacon/storage.go b/portalnetwork/beacon/storage.go index a49a6135c0..f17d7a17c7 100644 --- a/portalnetwork/beacon/storage.go +++ b/portalnetwork/beacon/storage.go @@ -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 } diff --git a/portalnetwork/history/storage.go b/portalnetwork/history/storage.go index 488d2ff362..7784809109 100644 --- a/portalnetwork/history/storage.go +++ b/portalnetwork/history/storage.go @@ -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 } diff --git a/portalnetwork/state/storage.go b/portalnetwork/state/storage.go index dae8a14d1d..2270d48ad6 100644 --- a/portalnetwork/state/storage.go +++ b/portalnetwork/state/storage.go @@ -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 } diff --git a/portalnetwork/state/storage_test.go b/portalnetwork/state/storage_test.go index 7212e14407..180eaa9b0a 100644 --- a/portalnetwork/state/storage_test.go +++ b/portalnetwork/state/storage_test.go @@ -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)