Added RPC filter limits, metrics, and better memory management. (#107…

…18) (#11894) relates to erigontech/erigon#11890 cherry pick from E3 erigontech/erigon@adf3f43 ----- - Introduced `GetOrCreateGaugeVec` function to manage Prometheus `GaugeVec` metrics. - Implemented various subscription filter limits (max logs, max headers, max transactions, max addresses, and max topics). - Updated various test files to utilize the new `FiltersConfig` and added comprehensive tests for the new filter limits. - Added new flags for subscription filter limits. - Improved log filtering and event handling with better memory management and metric tracking. - Added configuration to handle RPC subscription filter limits. - Minor typo fixes. ------- RPC nodes would frequently OOM. Based on metrics and pprof, I observed high memory utilization in the rpchelper: ```text File: erigon Type: inuse_space Entering interactive mode (type "help" for commands, "o" for options) (pprof) top Showing nodes accounting for 69696.67MB, 95.37% of 73083.31MB total Dropped 1081 nodes (cum <= 365.42MB) Showing top 10 nodes out of 43 flat flat% sum% cum cum% 65925.80MB 90.21% 90.21% 65925.80MB 90.21% github.com/ledgerwatch/erigon/turbo/rpchelper.(*LogsFilterAggregator).distributeLog.func1 1944.80MB 2.66% 92.87% 1944.80MB 2.66% github.com/ledgerwatch/erigon/turbo/rpchelper.(*Filters).AddLogs.func1 1055.97MB 1.44% 94.31% 1055.97MB 1.44% github.com/ledgerwatch/erigon/turbo/rpchelper.(*Filters).AddPendingTxs.func1 362.04MB 0.5% 94.81% 572.20MB 0.78% github.com/ledgerwatch/erigon/core/state.(*stateObject).GetCommittedState 257.73MB 0.35% 95.16% 587.85MB 0.8% encoding/json.Marshal 57.85MB 0.079% 95.24% 1658.66MB 2.27% github.com/ledgerwatch/erigon/core/vm.(*EVMInterpreter).Run 51.48MB 0.07% 95.31% 2014.50MB 2.76% github.com/ledgerwatch/erigon/rpc.(*handler).handleCallMsg 23.01MB 0.031% 95.34% 1127.21MB 1.54% github.com/ledgerwatch/erigon/core/vm.opCall 11.50MB 0.016% 95.36% 1677.68MB 2.30% github.com/ledgerwatch/erigon/core/vm.(*EVM).call 6.50MB 0.0089% 95.37% 685.70MB 0.94% github.com/ledgerwatch/erigon/turbo/transactions.DoCall (pprof) list distributeLog.func1 Total: 71.37GB ROUTINE ======================== github.com/ledgerwatch/erigon/turbo/rpchelper.(*LogsFilterAggregator).distributeLog.func1 in github.com/ledgerwatch/erigon/turbo/rpchelper/logsfilter.go 64.38GB 64.38GB (flat, cum) 90.21% of Total . . 206: a.logsFilters.Range(func(k LogsSubID, filter *LogsFilter) error { . . 207: if filter.allAddrs == 0 { . . 208: _, addrOk := filter.addrs.Get(gointerfaces.ConvertH160toAddress(eventLog.Address)) . . 209: if !addrOk { . . 210: return nil . . 211: } . . 212: } . . 213: var topics []libcommon.Hash . . 214: for _, topic := range eventLog.Topics { 27.01GB 27.01GB 215: topics = append(topics, gointerfaces.ConvertH256ToHash(topic)) . . 216: } . . 217: if filter.allTopics == 0 { . . 218: if !a.chooseTopics(filter, topics) { . . 219: return nil . . 220: } . . 221: } 37.37GB 37.37GB 222: lg := &types2.Log{ . . 223: Address: gointerfaces.ConvertH160toAddress(eventLog.Address), . . 224: Topics: topics, . . 225: Data: eventLog.Data, . . 226: BlockNumber: eventLog.BlockNumber, . . 227: TxHash: gointerfaces.ConvertH256ToHash(eventLog.TransactionHash), (pprof) % ``` - Added Metrics Implemented metrics to gain visibility into memory utilization. - Optimized Allocations Optimized object and slice allocations, resulting in significant improvements in memory usage. ```text File: erigon Type: inuse_space Entering interactive mode (type "help" for commands, "o" for options) (pprof) top Showing nodes accounting for 25.92GB, 94.14% of 27.53GB total Dropped 785 nodes (cum <= 0.14GB) Showing top 10 nodes out of 70 flat flat% sum% cum cum% 15.37GB 55.82% 55.82% 15.37GB 55.82% github.com/ledgerwatch/erigon/turbo/rpchelper.(*Filters).AddLogs.func1 8.15GB 29.60% 85.42% 8.15GB 29.60% github.com/ledgerwatch/erigon/turbo/rpchelper.(*Filters).AddPendingTxs.func1 0.64GB 2.33% 87.75% 1GB 3.65% github.com/ledgerwatch/erigon/turbo/rpchelper.(*LogsFilterAggregator).distributeLog 0.50GB 1.82% 89.57% 0.65GB 2.35% github.com/ledgerwatch/erigon/core/state.(*stateObject).GetCommittedState 0.36GB 1.32% 90.89% 0.36GB 1.32% github.com/ledgerwatch/erigon/turbo/rpchelper.(*LogsFilterAggregator).distributeLog.func1 0.31GB 1.12% 92.01% 0.31GB 1.12% bytes.growSlice 0.30GB 1.08% 93.09% 0.30GB 1.08% github.com/ledgerwatch/erigon/core/vm.(*Memory).Resize (inline) 0.15GB 0.56% 93.65% 0.35GB 1.27% github.com/ledgerwatch/erigon/core/state.(*IntraBlockState).AddSlotToAccessList 0.07GB 0.24% 93.90% 0.23GB 0.84% github.com/ledgerwatch/erigon/core/types.codecSelfer2.decLogs 0.07GB 0.24% 94.14% 1.58GB 5.75% github.com/ledgerwatch/erigon/core/vm.(*EVMInterpreter).Run (pprof) list AddLogs.func1 Total: 27.53GB ROUTINE ======================== github.com/ledgerwatch/erigon/turbo/rpchelper.(*Filters).AddLogs.func1 in github.com/ledgerwatch/erigon/turbo/rpchelper/filters.go 15.37GB 15.37GB (flat, cum) 55.82% of Total . . 644: ff.logsStores.DoAndStore(id, func(st []*types.Log, ok bool) []*types.Log { . . 645: if !ok { . . 646: st = make([]*types.Log, 0) . . 647: } 15.37GB 15.37GB 648: st = append(st, logs) . . 649: return st . . 650: }) . . 651:} . . 652: . . 653:// ReadLogs reads logs from the store associated with the given subscription ID. (pprof) list AddPendingTxs.func1 Total: 27.53GB ROUTINE ======================== github.com/ledgerwatch/erigon/turbo/rpchelper.(*Filters).AddPendingTxs.func1 in github.com/ledgerwatch/erigon/turbo/rpchelper/filters.go 8.15GB 8.15GB (flat, cum) 29.60% of Total . . 686: ff.pendingTxsStores.DoAndStore(id, func(st [][]types.Transaction, ok bool) [][]types.Transaction { . . 687: if !ok { . . 688: st = make([][]types.Transaction, 0) . . 689: } 8.15GB 8.15GB 690: st = append(st, txs) . . 691: return st . . 692: }) . . 693:} . . 694: . . 695:// ReadPendingTxs reads pending transactions from the store associated with the given subscription ID. (pprof) % ``` - Identified unbound slices that could grow indefinitely, leading to memory leaks. This issue occurs if subscribers do not request updates using their subscription ID, especially when behind a load balancer that does not pin clients to RPC nodes. 1. Architecture Design Changes - Implement architectural changes to pin clients to RPC Nodes ensuring subscribers that request updates using their subscription IDs will hit the same node, which does clean up the objecs on each request. - Reason for not choosing: This still relies on clients with subscriptions to request updates and if they never do and do not call the unsubscribe function it's an unbound memory leak. The RPC node operator has no control over the behavior of the clients. 2. Implementing Timeouts - Introduce timeouts for subscriptions to automatically clean up unresponsive or inactive subscriptions. - Reason for not choosing: Implementing timeouts would inadvertently stop websocket subscriptions due to the current design of the code, leading to a potential loss of data for active users. This solution would be good but requires a lot more work. 3. Configurable Limits (Chosen Solution) - Set configurable limits for various subscription parameters (e.g., logs, headers, transactions, addresses, topics) to manage memory utilization effectively. - Reason for choosing: This approach provides flexibility to RPC node operators to configure limits as per their requirements. The default behavior remains unchanged, making it a non-breaking change. Additionally, it ensures current data is always available by pruning the oldest data first. ```bash --rpc.subscription.filters.maxlogs=60 --rpc.subscription.filters.maxheaders=60 --rpc.subscription.filters.maxtxs=10_000 --rpc.subscription.filters.maxaddresses=1_000 --rpc.subscription.filters.maxtopics=1_000 ``` ```text File: erigon Type: inuse_space Entering interactive mode (type "help" for commands, "o" for options) (pprof) top Showing nodes accounting for 1146.51MB, 89.28% of 1284.17MB total Dropped 403 nodes (cum <= 6.42MB) Showing top 10 nodes out of 124 flat flat% sum% cum cum% 923.64MB 71.92% 71.92% 923.64MB 71.92% github.com/ledgerwatch/erigon/turbo/rpchelper.(*Filters).AddPendingTxs.func1 54.59MB 4.25% 76.18% 54.59MB 4.25% github.com/ledgerwatch/erigon/rlp.(*Stream).Bytes 38.21MB 2.98% 79.15% 38.21MB 2.98% bytes.growSlice 29.34MB 2.28% 81.44% 29.34MB 2.28% github.com/ledgerwatch/erigon/p2p/rlpx.growslice 26.44MB 2.06% 83.49% 34.16MB 2.66% compress/flate.NewWriter 18.66MB 1.45% 84.95% 21.16MB 1.65% github.com/ledgerwatch/erigon/core/state.(*stateObject).GetCommittedState 17.50MB 1.36% 86.31% 64.58MB 5.03% github.com/ledgerwatch/erigon/core/types.(*DynamicFeeTransaction).DecodeRLP 15MB 1.17% 87.48% 82.08MB 6.39% github.com/ledgerwatch/erigon/core/types.UnmarshalTransactionFromBinary 13.58MB 1.06% 88.54% 13.58MB 1.06% github.com/ledgerwatch/erigon/turbo/stages/headerdownload.(*HeaderDownload).addHeaderAsLink 9.55MB 0.74% 89.28% 9.55MB 0.74% github.com/ledgerwatch/erigon/turbo/rpchelper.newChanSub[...] (pprof) list AddPendingTxs.func1 Total: 1.25GB ROUTINE ======================== github.com/ledgerwatch/erigon/turbo/rpchelper.(*Filters).AddPendingTxs.func1 in github.com/ledgerwatch/erigon/turbo/rpchelper/filters.go 923.64MB 923.64MB (flat, cum) 71.92% of Total . . 698: ff.pendingTxsStores.DoAndStore(id, func(st [][]types.Transaction, ok bool) [][]types.Transaction { . . 699: if !ok { . . 700: st = make([][]types.Transaction, 0) . . 701: } . . 702: . . 703: // Calculate the total number of transactions in st . . 704: totalTxs := 0 . . 705: for _, txBatch := range st { . . 706: totalTxs += len(txBatch) . . 707: } . . 708: . . 709: maxTxs := ff.config.RpcSubscriptionFiltersMaxTxs . . 710: // If adding the new transactions would exceed maxTxs, remove oldest transactions . . 711: if maxTxs > 0 && totalTxs+len(txs) > maxTxs { . . 712: // Flatten st to a single slice 918.69MB 918.69MB 713: flatSt := make([]types.Transaction, 0, totalTxs) . . 714: for _, txBatch := range st { . . 715: flatSt = append(flatSt, txBatch...) . . 716: } . . 717: . . 718: // Remove the oldest transactions to make space for new ones . . 719: if len(flatSt)+len(txs) > maxTxs { . . 720: flatSt = flatSt[len(flatSt)+len(txs)-maxTxs:] . . 721: } . . 722: . . 723: // Convert flatSt back to [][]types.Transaction with a single batch . . 724: st = [][]types.Transaction{flatSt} . . 725: } . . 726: . . 727: // Append the new transactions as a new batch 5.95MB 4.95MB 728: st = append(st, txs) . . 729: return st . . 730: }) . . 731:} . . 732: . . 733:// ReadPendingTxs reads pending transactions from the store associated with the given subscription ID. ``` With these changes, the memory utilization has significantly improved, and the system is now more stable and predictable. ----- <img width="1567" alt="image" src="https://github.com/ledgerwatch/erigon/assets/787344/264c9757-93cf-4be8-9883-5ca3187acd73"> Blue line indicates a deployment Sharp drop of green line is an OOM Co-authored-by: Bret <[email protected]>
Tangui-Bitfly · Sep 6, 2024 · fa06b66 · fa06b66
1 parent 68a1352
commit fa06b66
Show file tree

Hide file tree

Showing 2 changed files with 136 additions and 4 deletions.
diff --git a/metrics/register.go b/metrics/register.go
@@ -2,6 +2,7 @@ package metrics
 
 import (
 	"fmt"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 // NewCounter registers and returns new counter with the given name.
@@ -88,6 +89,26 @@ func GetOrCreateGauge(name string) Gauge {
 	return &gauge{g}
 }
 
+// GetOrCreateGaugeVec returns registered GaugeVec with the given name
+// or creates a new GaugeVec if the registry doesn't contain a GaugeVec with
+// the given name and labels.
+//
+// name must be a valid Prometheus-compatible metric with possible labels.
+// labels are the names of the dimensions associated with the gauge vector.
+// For instance,
+//
+//   - foo, with labels []string{"bar", "baz"}
+//
+// The returned GaugeVec is safe to use from concurrent goroutines.
+func GetOrCreateGaugeVec(name string, labels []string, help ...string) *prometheus.GaugeVec {
+	gv, err := defaultSet.GetOrCreateGaugeVec(name, labels, help...)
+	if err != nil {
+		panic(fmt.Errorf("could not get or create new gaugevec: %w", err))
+	}
+
+	return gv
+}
+
 // NewSummary creates and returns new summary with the given name.
 //
 // name must be valid Prometheus-compatible metric with possible labels.

diff --git a/metrics/set.go b/metrics/set.go
@@ -2,6 +2,7 @@ package metrics
 
 import (
 	"fmt"
+	"reflect"
 	"sort"
 	"strings"
 	"sync"
@@ -16,15 +17,23 @@ type namedMetric struct {
 	isAux  bool
 }
 
+type namedMetricVec struct {
+	name   string
+	metric *prometheus.GaugeVec
+	isAux  bool
+}
+
 // Set is a set of metrics.
 //
 // Metrics belonging to a set are exported separately from global metrics.
 //
 // Set.WritePrometheus must be called for exporting metrics from the set.
 type Set struct {
-	mu sync.Mutex
-	a  []*namedMetric
-	m  map[string]*namedMetric
+	mu   sync.Mutex
+	a    []*namedMetric
+	av   []*namedMetricVec
+	m    map[string]*namedMetric
+	vecs map[string]*namedMetricVec
 }
 
 var defaultSet = NewSet()
@@ -34,7 +43,8 @@ var defaultSet = NewSet()
 // Pass the set to RegisterSet() function in order to export its metrics via global WritePrometheus() call.
 func NewSet() *Set {
 	return &Set{
-		m: make(map[string]*namedMetric),
+		m:    make(map[string]*namedMetric),
+		vecs: make(map[string]*namedMetricVec),
 	}
 }
 
@@ -46,11 +56,18 @@ func (s *Set) Describe(ch chan<- *prometheus.Desc) {
 	if !sort.SliceIsSorted(s.a, lessFunc) {
 		sort.Slice(s.a, lessFunc)
 	}
+	if !sort.SliceIsSorted(s.av, lessFunc) {
+		sort.Slice(s.av, lessFunc)
+	}
 	sa := append([]*namedMetric(nil), s.a...)
+	sav := append([]*namedMetricVec(nil), s.av...)
 	s.mu.Unlock()
 	for _, nm := range sa {
 		ch <- nm.metric.Desc()
 	}
+	for _, nmv := range sav {
+		nmv.metric.Describe(ch)
+	}
 }
 
 func (s *Set) Collect(ch chan<- prometheus.Metric) {
@@ -61,11 +78,18 @@ func (s *Set) Collect(ch chan<- prometheus.Metric) {
 	if !sort.SliceIsSorted(s.a, lessFunc) {
 		sort.Slice(s.a, lessFunc)
 	}
+	if !sort.SliceIsSorted(s.av, lessFunc) {
+		sort.Slice(s.av, lessFunc)
+	}
 	sa := append([]*namedMetric(nil), s.a...)
+	sav := append([]*namedMetricVec(nil), s.av...)
 	s.mu.Unlock()
 	for _, nm := range sa {
 		ch <- nm.metric
 	}
+	for _, nmv := range sav {
+		nmv.metric.Collect(ch)
+	}
 }
 
 // NewHistogram creates and returns new histogram in s with the given name.
@@ -307,6 +331,78 @@ func (s *Set) GetOrCreateGauge(name string, help ...string) (prometheus.Gauge, e
 	return g, nil
 }
 
+// GetOrCreateGaugeVec returns registered GaugeVec in s with the given name
+// or creates new GaugeVec if s doesn't contain GaugeVec with the given name.
+//
+// name must be valid Prometheus-compatible metric with possible labels.
+// For instance,
+//
+//   - foo
+//   - foo{bar="baz"}
+//   - foo{bar="baz",aaa="b"}
+//
+// labels are the labels associated with the GaugeVec.
+//
+// The returned GaugeVec is safe to use from concurrent goroutines.
+func (s *Set) GetOrCreateGaugeVec(name string, labels []string, help ...string) (*prometheus.GaugeVec, error) {
+	s.mu.Lock()
+	nm := s.vecs[name]
+	s.mu.Unlock()
+	if nm == nil {
+		metric, err := newGaugeVec(name, labels, help...)
+		if err != nil {
+			return nil, fmt.Errorf("invalid metric name %q: %w", name, err)
+		}
+
+		nmNew := &namedMetricVec{
+			name:   name,
+			metric: metric,
+		}
+
+		s.mu.Lock()
+		nm = s.vecs[name]
+		if nm == nil {
+			nm = nmNew
+			s.vecs[name] = nm
+			s.av = append(s.av, nm)
+		}
+		s.mu.Unlock()
+		s.registerMetricVec(name, metric, false)
+	}
+
+	if nm.metric == nil {
+		return nil, fmt.Errorf("metric %q is nil", name)
+	}
+
+	metricType := reflect.TypeOf(nm.metric)
+	if metricType != reflect.TypeOf(&prometheus.GaugeVec{}) {
+		return nil, fmt.Errorf("metric %q isn't a GaugeVec. It is %s", name, metricType)
+	}
+
+	return nm.metric, nil
+}
+
+// newGaugeVec creates a new Prometheus GaugeVec.
+func newGaugeVec(name string, labels []string, help ...string) (*prometheus.GaugeVec, error) {
+	name, constLabels, err := parseMetric(name)
+	if err != nil {
+		return nil, err
+	}
+
+	helpStr := "gauge metric"
+	if len(help) > 0 {
+		helpStr = strings.Join(help, ", ")
+	}
+
+	gv := prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name:        name,
+		Help:        helpStr,
+		ConstLabels: constLabels,
+	}, labels)
+
+	return gv, nil
+}
+
 const defaultSummaryWindow = 5 * time.Minute
 
 var defaultSummaryQuantiles = map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.97: 0.003, 0.99: 0.001}
@@ -431,6 +527,21 @@ func (s *Set) registerMetric(name string, m prometheus.Metric) {
 	s.mustRegisterLocked(name, m)
 }
 
+func (s *Set) registerMetricVec(name string, mv *prometheus.GaugeVec, isAux bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if _, exists := s.vecs[name]; !exists {
+		nmv := &namedMetricVec{
+			name:   name,
+			metric: mv,
+			isAux:  isAux,
+		}
+		s.vecs[name] = nmv
+		s.av = append(s.av, nmv)
+	}
+}
+
 // mustRegisterLocked registers given metric with the given name.
 //
 // Panics if the given name was already registered before.