From 612895de901dde15eabc4274ba421d6f1e2bf1b7 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 1/3] 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 6977180a4dd9..ea2d918e6daf 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 000000000000..1d4cb3bcd9d1 --- /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 a49a6135c012..f17d7a17c74c 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 488d2ff3622c..778480910940 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 dae8a14d1da1..2270d48ad6eb 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 7212e14407f7..180eaa9b0ada 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) From d19c1f3d891f6ed9730f45bb43e75cf81e4ccaed Mon Sep 17 00:00:00 2001 From: Rafael Sampaio <5679073+r4f4ss@users.noreply.github.com> Date: Sun, 27 Oct 2024 23:43:28 -0300 Subject: [PATCH 2/3] fix ci tests: unnecessary conversion --- portalnetwork/beacon/storage.go | 2 +- portalnetwork/state/storage.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/portalnetwork/beacon/storage.go b/portalnetwork/beacon/storage.go index f17d7a17c74c..33f0a458afa9 100644 --- a/portalnetwork/beacon/storage.go +++ b/portalnetwork/beacon/storage.go @@ -89,7 +89,7 @@ func NewBeaconStorage(config storage.PortalStorageConfig) (storage.ContentStorag } else { q.Scan(resStr) } - contentStorageUsage.Update(int64(*resStr)) + contentStorageUsage.Update(*resStr) } return bs, nil } diff --git a/portalnetwork/state/storage.go b/portalnetwork/state/storage.go index 2270d48ad6eb..a2c771e8854b 100644 --- a/portalnetwork/state/storage.go +++ b/portalnetwork/state/storage.go @@ -81,7 +81,7 @@ func NewStateStorage(store storage.ContentStorage, db *sql.DB) *StateStorage { } else { q.Scan(resStr) } - contentStorageUsage.Update(int64(*resStr)) + contentStorageUsage.Update(*resStr) } return storage From 1d8bba1481d220f072e60af142b88f7ba2a430f9 Mon Sep 17 00:00:00 2001 From: Rafael Sampaio <5679073+r4f4ss@users.noreply.github.com> Date: Mon, 28 Oct 2024 15:56:21 -0300 Subject: [PATCH 3/3] change log levels and extract duplicate logic --- cmd/shisui/main.go | 1 + metrics/portal_metrics.go | 71 ++++++++++++++++++++++++++++++++ portalnetwork/beacon/storage.go | 62 +++++----------------------- portalnetwork/history/storage.go | 48 +++++++-------------- portalnetwork/state/storage.go | 66 ++++++----------------------- 5 files changed, 110 insertions(+), 138 deletions(-) diff --git a/cmd/shisui/main.go b/cmd/shisui/main.go index ea2d918e6daf..35e754c09c00 100644 --- a/cmd/shisui/main.go +++ b/cmd/shisui/main.go @@ -419,6 +419,7 @@ func initBeacon(config Config, server *rpc.Server, conn discover.UDPConn, localN DB: sqlDb, NodeId: localNode.ID(), Spec: configs.Mainnet, + NetworkName: portalwire.Beacon.Name(), }) if err != nil { return nil, err diff --git a/metrics/portal_metrics.go b/metrics/portal_metrics.go index 1d4cb3bcd9d1..8d524ffdddf3 100644 --- a/metrics/portal_metrics.go +++ b/metrics/portal_metrics.go @@ -1,9 +1,12 @@ package metrics import ( + "database/sql" + "errors" "os" "path" "slices" + "strings" "time" "github.com/ethereum/go-ethereum/log" @@ -17,6 +20,17 @@ type networkFileMetric struct { network string } +type PortalStorageMetrics struct { + RadiusRatio GaugeFloat64 + EntriesCount Gauge + ContentStorageUsage Gauge +} + +const ( + countEntrySql = "SELECT COUNT(1) FROM kvstore;" + contentStorageUsageSql = "SELECT SUM( length(value) ) FROM kvstore;" +) + // 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 @@ -80,3 +94,60 @@ func CollectPortalMetrics(refresh time.Duration, networks []string, dataDir stri time.Sleep(refresh) } } + +func NewPortalStorageMetrics(network string, db *sql.DB) (*PortalStorageMetrics, error) { + if !Enabled { + return nil, nil + } + + if network != portalwire.History.Name() && network != portalwire.Beacon.Name() && network != portalwire.State.Name() { + log.Debug("Unknow network for metrics", "network", network) + return nil, errors.New("unknow network for metrics") + } + + var countSql string + var contentSql string + if network == portalwire.Beacon.Name() { + countSql = strings.Replace(countEntrySql, "kvstore", "beacon", 1) + contentSql = strings.Replace(contentStorageUsageSql, "kvstore", "beacon", 1) + contentSql = strings.Replace(contentSql, "value", "content_value", 1) + } else { + countSql = countEntrySql + contentSql = contentStorageUsageSql + } + + storageMetrics := &PortalStorageMetrics{} + + storageMetrics.RadiusRatio = NewRegisteredGaugeFloat64("portal/"+network+"/radius_ratio", nil) + storageMetrics.RadiusRatio.Update(1) + + storageMetrics.EntriesCount = NewRegisteredGauge("portal/"+network+"/entry_count", nil) + log.Debug("Counting entities in " + network + " storage for metrics") + var res *int64 = new(int64) + q := db.QueryRow(countSql) + if q.Err() == sql.ErrNoRows { + storageMetrics.EntriesCount.Update(0) + } else if q.Err() != nil { + log.Error("Querry execution error", "network", network, "metric", "entry_count", "err", q.Err()) + return nil, q.Err() + } else { + q.Scan(res) + storageMetrics.EntriesCount.Update(*res) + } + + storageMetrics.ContentStorageUsage = NewRegisteredGauge("portal/"+network+"/content_storage", nil) + log.Debug("Counting storage usage (bytes) in " + network + " for metrics") + var res2 *int64 = new(int64) + q2 := db.QueryRow(contentSql) + if q2.Err() == sql.ErrNoRows { + storageMetrics.ContentStorageUsage.Update(0) + } else if q2.Err() != nil { + log.Error("Querry execution error", "network", network, "metric", "entry_count", "err", q2.Err()) + return nil, q2.Err() + } else { + q2.Scan(res2) + storageMetrics.ContentStorageUsage.Update(*res2) + } + + return storageMetrics, nil +} diff --git a/portalnetwork/beacon/storage.go b/portalnetwork/beacon/storage.go index 33f0a458afa9..ff90835313d8 100644 --- a/portalnetwork/beacon/storage.go +++ b/portalnetwork/beacon/storage.go @@ -16,17 +16,6 @@ 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 @@ -35,6 +24,8 @@ type BeaconStorage struct { cache *beaconStorageCache } +var portalStorageMetrics *metrics.PortalStorageMetrics + type beaconStorageCache struct { OptimisticUpdate []byte FinalityUpdate []byte @@ -53,44 +44,13 @@ 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(*resStr) + var err error + portalStorageMetrics, err = metrics.NewPortalStorageMetrics(config.NetworkName, config.DB) + if err != nil { + return nil, err } + return bs, nil } @@ -226,8 +186,8 @@ func (bs *BeaconStorage) putContentValue(contentId, contentKey, value []byte) er 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))) + portalStorageMetrics.EntriesCount.Inc(1) + portalStorageMetrics.ContentStorageUsage.Inc(int64(len(value))) } return err } @@ -235,8 +195,8 @@ func (bs *BeaconStorage) putContentValue(contentId, contentKey, value []byte) er 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))) + portalStorageMetrics.EntriesCount.Inc(1) + portalStorageMetrics.ContentStorageUsage.Inc(int64(len(value))) } return err } diff --git a/portalnetwork/history/storage.go b/portalnetwork/history/storage.go index 778480910940..ee43fb246eb5 100644 --- a/portalnetwork/history/storage.go +++ b/portalnetwork/history/storage.go @@ -21,12 +21,6 @@ import ( "github.com/mattn/go-sqlite3" ) -var ( - radiusRatio metrics.GaugeFloat64 - entriesCount metrics.Gauge - contentStorageUsage metrics.Gauge -) - const ( sqliteName = "history.sqlite" contentDeletionFraction = 0.05 // 5% of the content will be deleted when the storage capacity is hit and radius gets adjusted. @@ -62,6 +56,8 @@ type ContentStorage struct { log log.Logger } +var portalStorageMetrics *metrics.PortalStorageMetrics + func xor(contentId, nodeId []byte) []byte { // length of contentId maybe not 32bytes padding := make([]byte, 32) @@ -125,25 +121,11 @@ 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 strings.ToLower(config.NetworkName) == "history" { + portalStorageMetrics, err = metrics.NewPortalStorageMetrics(config.NetworkName, config.DB) if err != nil { - log.Debug("Querry execution error", "network", config.NetworkName, "metric", "content_storage", "err", err) + return nil, err } - contentStorageUsage.Update(int64(str)) } return hs, err @@ -217,8 +199,8 @@ func (p *ContentStorage) put(contentId []byte, content []byte) PutResult { } if metrics.Enabled { - entriesCount.Inc(1) - contentStorageUsage.Inc(int64(len(content))) + portalStorageMetrics.EntriesCount.Inc(1) + portalStorageMetrics.ContentStorageUsage.Inc(int64(len(content))) } return PutResult{} } @@ -387,7 +369,7 @@ func (p *ContentStorage) EstimateNewRadius(currentRadius *uint256.Int) (*uint256 newRadius := new(uint256.Int).Div(currentRadius, uint256.MustFromBig(bigFormat)) newRadius.Mul(newRadius, uint256.NewInt(100)) newRadius.Mod(newRadius, storage.MaxDistance) - radiusRatio.Update(newRadius.Float64() / 100) + portalStorageMetrics.RadiusRatio.Update(newRadius.Float64() / 100) } return new(uint256.Int).Div(currentRadius, uint256.MustFromBig(bigFormat)), nil } @@ -451,7 +433,7 @@ func (p *ContentStorage) deleteContentFraction(fraction float64) (deleteCount in if metrics.Enabled { dis.Mul(dis, uint256.NewInt(100)) dis.Mod(dis, storage.MaxDistance) - radiusRatio.Update(dis.Float64() / 100) + portalStorageMetrics.RadiusRatio.Update(dis.Float64() / 100) } } // row must close first, or database is locked @@ -475,8 +457,8 @@ func (p *ContentStorage) del(contentId []byte) error { } _, err = p.delStmt.Exec(contentId) if metrics.Enabled && err != nil { - entriesCount.Dec(1) - contentStorageUsage.Dec(int64(sizeDel)) + portalStorageMetrics.EntriesCount.Dec(1) + portalStorageMetrics.ContentStorageUsage.Dec(int64(sizeDel)) } return err } @@ -499,8 +481,8 @@ func (p *ContentStorage) batchDel(ids [][]byte) error { // delete items _, err = p.sqliteDB.Exec(query, args...) if metrics.Enabled && err != nil { - entriesCount.Dec(int64(len(args))) - contentStorageUsage.Dec(int64(sizeDel)) + portalStorageMetrics.EntriesCount.Dec(int64(len(args))) + portalStorageMetrics.ContentStorageUsage.Dec(int64(sizeDel)) } return err } @@ -528,8 +510,8 @@ func (p *ContentStorage) deleteContentOutOfRadius(radius *uint256.Int) error { count, err := res.RowsAffected() p.log.Trace("delete items", "count", count) if metrics.Enabled && err != nil { - entriesCount.Dec(count) - contentStorageUsage.Dec(int64(sizeDel)) + portalStorageMetrics.EntriesCount.Dec(count) + portalStorageMetrics.ContentStorageUsage.Dec(int64(sizeDel)) } return err } diff --git a/portalnetwork/state/storage.go b/portalnetwork/state/storage.go index a2c771e8854b..466c1707d6a2 100644 --- a/portalnetwork/state/storage.go +++ b/portalnetwork/state/storage.go @@ -14,17 +14,6 @@ import ( "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[:] @@ -38,6 +27,8 @@ type StateStorage struct { log log.Logger } +var portalStorageMetrics *metrics.PortalStorageMetrics + func NewStateStorage(store storage.ContentStorage, db *sql.DB) *StateStorage { storage := &StateStorage{ store: store, @@ -45,43 +36,10 @@ func NewStateStorage(store storage.ContentStorage, db *sql.DB) *StateStorage { 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(*resStr) + var err error + portalStorageMetrics, err = metrics.NewPortalStorageMetrics("state", db) + if err != nil { + return nil } return storage @@ -141,8 +99,8 @@ func (s *StateStorage) putAccountTrieNode(contentKey []byte, contentId []byte, c 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))) + portalStorageMetrics.EntriesCount.Inc(1) + portalStorageMetrics.ContentStorageUsage.Inc(int64(len(content))) } return nil } @@ -178,8 +136,8 @@ func (s *StateStorage) putContractStorageTrieNode(contentKey []byte, contentId [ 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))) + portalStorageMetrics.EntriesCount.Inc(1) + portalStorageMetrics.ContentStorageUsage.Inc(int64(len(content))) } return nil } @@ -211,8 +169,8 @@ func (s *StateStorage) putContractBytecode(contentKey []byte, contentId []byte, 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))) + portalStorageMetrics.EntriesCount.Inc(1) + portalStorageMetrics.ContentStorageUsage.Inc(int64(len(content))) } return nil }