From 3ade355f6f3956b1ff1ad173c1ff257fdbe3ee63 Mon Sep 17 00:00:00 2001 From: Martin Holst Swende Date: Wed, 27 Nov 2024 04:24:01 +0100 Subject: [PATCH] all: different metrics-api, enable from flags --- cmd/geth/chaincmd.go | 9 ++-- cmd/geth/config.go | 5 ++ cmd/geth/main.go | 7 --- cmd/utils/flags.go | 93 ++++++++++++++++------------------ core/state/trie_prefetcher.go | 2 +- core/txpool/txpool.go | 4 +- eth/handler.go | 2 +- eth/protocols/eth/handler.go | 2 +- eth/protocols/eth/handshake.go | 2 +- eth/protocols/snap/handler.go | 2 +- metrics/metrics.go | 54 +++++++------------- metrics/resetting_timer.go | 2 +- p2p/discover/metrics.go | 2 +- p2p/discover/table.go | 4 +- p2p/metrics.go | 4 +- p2p/peer.go | 2 +- p2p/tracker/tracker.go | 4 +- p2p/transport.go | 2 +- 18 files changed, 90 insertions(+), 112 deletions(-) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index f6dc1cf4bf..913e72faf2 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -282,14 +282,15 @@ func importChain(ctx *cli.Context) error { if ctx.Args().Len() < 1 { utils.Fatalf("This command requires an argument.") } + stack, cfg := makeConfigNode(ctx) + defer stack.Close() + // Start metrics export if enabled - utils.SetupMetrics(ctx) + utils.SetupMetrics(&cfg.Metrics) + // Start system runtime metrics collection go metrics.CollectProcessMetrics(3 * time.Second) - stack, _ := makeConfigNode(ctx) - defer stack.Close() - chain, db := utils.MakeChain(ctx, stack, false) defer db.Close() diff --git a/cmd/geth/config.go b/cmd/geth/config.go index 17ed9fb606..e1dc0fda31 100644 --- a/cmd/geth/config.go +++ b/cmd/geth/config.go @@ -25,6 +25,7 @@ import ( "runtime" "slices" "strings" + "time" "unicode" "github.com/ethereum/go-ethereum/accounts" @@ -192,6 +193,10 @@ func makeFullNode(ctx *cli.Context) *node.Node { cfg.Eth.OverrideVerkle = &v } + // Start metrics export if enabled + utils.SetupMetrics(&cfg.Metrics) + go metrics.CollectProcessMetrics(3 * time.Second) + backend, eth := utils.RegisterEthService(stack, &cfg.Eth) // Create gauge with geth system and build information diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 1527e180fb..9d9256862b 100644 --- a/cmd/geth/main.go +++ b/cmd/geth/main.go @@ -34,7 +34,6 @@ import ( "github.com/ethereum/go-ethereum/internal/debug" "github.com/ethereum/go-ethereum/internal/flags" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/metrics" "github.com/ethereum/go-ethereum/node" "go.uber.org/automaxprocs/maxprocs" @@ -325,12 +324,6 @@ func prepare(ctx *cli.Context) { ctx.Set(utils.CacheFlag.Name, strconv.Itoa(4096)) } } - - // Start metrics export if enabled - utils.SetupMetrics(ctx) - - // Start system runtime metrics collection - go metrics.CollectProcessMetrics(3 * time.Second) } // geth is the main entry point into the system if no special subcommand is run. diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index e635dd89c3..712729ec64 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -1969,64 +1969,61 @@ func RegisterFullSyncTester(stack *node.Node, eth *eth.Ethereum, target common.H log.Info("Registered full-sync tester", "hash", target) } -func SetupMetrics(ctx *cli.Context) { - if metrics.Enabled { - log.Info("Enabling metrics collection") +func SetupMetrics(cfg *metrics.Config) { + if !cfg.Enabled { + return + } + metrics.Init(true) + log.Info("Enabling metrics collection") + var ( + enableExport = cfg.EnableInfluxDB + enableExportV2 = cfg.EnableInfluxDBV2 + ) + if enableExport && enableExportV2 { + Fatalf("Flags %v can't be used at the same time", strings.Join([]string{MetricsEnableInfluxDBFlag.Name, MetricsEnableInfluxDBV2Flag.Name}, ", ")) + } + if enableExport || enableExportV2 { + v1FlagIsSet := cfg.InfluxDBUsername != "" || cfg.InfluxDBPassword != "" + v2FlagIsSet := cfg.InfluxDBToken != "" || cfg.InfluxDBOrganization != "" || cfg.InfluxDBBucket != "" - var ( - enableExport = ctx.Bool(MetricsEnableInfluxDBFlag.Name) - enableExportV2 = ctx.Bool(MetricsEnableInfluxDBV2Flag.Name) - ) - - if enableExport || enableExportV2 { - CheckExclusive(ctx, MetricsEnableInfluxDBFlag, MetricsEnableInfluxDBV2Flag) - - v1FlagIsSet := ctx.IsSet(MetricsInfluxDBUsernameFlag.Name) || - ctx.IsSet(MetricsInfluxDBPasswordFlag.Name) - - v2FlagIsSet := ctx.IsSet(MetricsInfluxDBTokenFlag.Name) || - ctx.IsSet(MetricsInfluxDBOrganizationFlag.Name) || - ctx.IsSet(MetricsInfluxDBBucketFlag.Name) - - if enableExport && v2FlagIsSet { - Fatalf("Flags --influxdb.metrics.organization, --influxdb.metrics.token, --influxdb.metrics.bucket are only available for influxdb-v2") - } else if enableExportV2 && v1FlagIsSet { - Fatalf("Flags --influxdb.metrics.username, --influxdb.metrics.password are only available for influxdb-v1") - } + if enableExport && v2FlagIsSet { + Fatalf("Flags --influxdb.metrics.organization, --influxdb.metrics.token, --influxdb.metrics.bucket are only available for influxdb-v2") + } else if enableExportV2 && v1FlagIsSet { + Fatalf("Flags --influxdb.metrics.username, --influxdb.metrics.password are only available for influxdb-v1") } + } - var ( - endpoint = ctx.String(MetricsInfluxDBEndpointFlag.Name) - database = ctx.String(MetricsInfluxDBDatabaseFlag.Name) - username = ctx.String(MetricsInfluxDBUsernameFlag.Name) - password = ctx.String(MetricsInfluxDBPasswordFlag.Name) + var ( + endpoint = cfg.InfluxDBEndpoint + database = cfg.InfluxDBDatabase + username = cfg.InfluxDBUsername + password = cfg.InfluxDBPassword - token = ctx.String(MetricsInfluxDBTokenFlag.Name) - bucket = ctx.String(MetricsInfluxDBBucketFlag.Name) - organization = ctx.String(MetricsInfluxDBOrganizationFlag.Name) - ) + token = cfg.InfluxDBToken + bucket = cfg.InfluxDBBucket + organization = cfg.InfluxDBOrganization + ) - if enableExport { - tagsMap := SplitTagsFlag(ctx.String(MetricsInfluxDBTagsFlag.Name)) + if enableExport { + tagsMap := SplitTagsFlag(cfg.InfluxDBTags) - log.Info("Enabling metrics export to InfluxDB") + log.Info("Enabling metrics export to InfluxDB") - go influxdb.InfluxDBWithTags(metrics.DefaultRegistry, 10*time.Second, endpoint, database, username, password, "geth.", tagsMap) - } else if enableExportV2 { - tagsMap := SplitTagsFlag(ctx.String(MetricsInfluxDBTagsFlag.Name)) + go influxdb.InfluxDBWithTags(metrics.DefaultRegistry, 10*time.Second, endpoint, database, username, password, "geth.", tagsMap) + } else if enableExportV2 { + tagsMap := SplitTagsFlag(cfg.InfluxDBTags) - log.Info("Enabling metrics export to InfluxDB (v2)") + log.Info("Enabling metrics export to InfluxDB (v2)") - go influxdb.InfluxDBV2WithTags(metrics.DefaultRegistry, 10*time.Second, endpoint, token, bucket, organization, "geth.", tagsMap) - } + go influxdb.InfluxDBV2WithTags(metrics.DefaultRegistry, 10*time.Second, endpoint, token, bucket, organization, "geth.", tagsMap) + } - if ctx.IsSet(MetricsHTTPFlag.Name) { - address := net.JoinHostPort(ctx.String(MetricsHTTPFlag.Name), fmt.Sprintf("%d", ctx.Int(MetricsPortFlag.Name))) - log.Info("Enabling stand-alone metrics HTTP endpoint", "address", address) - exp.Setup(address) - } else if ctx.IsSet(MetricsPortFlag.Name) { - log.Warn(fmt.Sprintf("--%s specified without --%s, metrics server will not start.", MetricsPortFlag.Name, MetricsHTTPFlag.Name)) - } + if cfg.HTTP != "" { + address := net.JoinHostPort(cfg.HTTP, fmt.Sprintf("%d", cfg.Port)) + log.Info("Enabling stand-alone metrics HTTP endpoint", "address", address) + exp.Setup(address) + } else if cfg.HTTP == "" && cfg.Port != 0 { + log.Warn(fmt.Sprintf("--%s specified without --%s, metrics server will not start.", MetricsPortFlag.Name, MetricsHTTPFlag.Name)) } } diff --git a/core/state/trie_prefetcher.go b/core/state/trie_prefetcher.go index 2ee6d9a79e..6f492cf9f2 100644 --- a/core/state/trie_prefetcher.go +++ b/core/state/trie_prefetcher.go @@ -111,7 +111,7 @@ func (p *triePrefetcher) terminate(async bool) { // report aggregates the pre-fetching and usage metrics and reports them. func (p *triePrefetcher) report() { - if !metrics.Enabled { + if !metrics.Enabled() { return } for _, fetcher := range p.fetchers { diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index ce455e806e..d54cf4968b 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -126,7 +126,7 @@ func (p *TxPool) reserver(id int, subpool SubPool) AddressReserver { return ErrAlreadyReserved } p.reservations[addr] = subpool - if metrics.Enabled { + if metrics.Enabled() { m := fmt.Sprintf("%s/%d", reservationsGaugeName, id) metrics.GetOrRegisterGauge(m, nil).Inc(1) } @@ -143,7 +143,7 @@ func (p *TxPool) reserver(id int, subpool SubPool) AddressReserver { return errors.New("address not owned") } delete(p.reservations, addr) - if metrics.Enabled { + if metrics.Enabled() { m := fmt.Sprintf("%s/%d", reservationsGaugeName, id) metrics.GetOrRegisterGauge(m, nil).Dec(1) } diff --git a/eth/handler.go b/eth/handler.go index b28081eef0..f550b94365 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -365,7 +365,7 @@ func (h *handler) runSnapExtension(peer *snap.Peer, handler snap.Handler) error defer h.decHandlers() if err := h.peers.registerSnapExtension(peer); err != nil { - if metrics.Enabled { + if metrics.Enabled() { if peer.Inbound() { snap.IngressRegistrationErrorMeter.Mark(1) } else { diff --git a/eth/protocols/eth/handler.go b/eth/protocols/eth/handler.go index dc32559c47..28db93a209 100644 --- a/eth/protocols/eth/handler.go +++ b/eth/protocols/eth/handler.go @@ -190,7 +190,7 @@ func handleMessage(backend Backend, peer *Peer) error { var handlers = eth68 // Track the amount of time it takes to serve the request and run the handler - if metrics.Enabled { + if metrics.Enabled() { h := fmt.Sprintf("%s/%s/%d/%#02x", p2p.HandleHistName, ProtocolName, peer.Version(), msg.Code) defer func(start time.Time) { sampler := func() metrics.Sample { diff --git a/eth/protocols/eth/handshake.go b/eth/protocols/eth/handshake.go index ea16a85b1e..68cf846925 100644 --- a/eth/protocols/eth/handshake.go +++ b/eth/protocols/eth/handshake.go @@ -112,7 +112,7 @@ func (p *Peer) readStatus(network uint64, status *StatusPacket, genesis common.H // markError registers the error with the corresponding metric. func markError(p *Peer, err error) { - if !metrics.Enabled { + if !metrics.Enabled() { return } m := meters.get(p.Inbound()) diff --git a/eth/protocols/snap/handler.go b/eth/protocols/snap/handler.go index d36f9621b1..374165ea65 100644 --- a/eth/protocols/snap/handler.go +++ b/eth/protocols/snap/handler.go @@ -132,7 +132,7 @@ func HandleMessage(backend Backend, peer *Peer) error { defer msg.Discard() start := time.Now() // Track the amount of time it takes to serve the request and run the handler - if metrics.Enabled { + if metrics.Enabled() { h := fmt.Sprintf("%s/%s/%d/%#02x", p2p.HandleHistName, ProtocolName, peer.Version(), msg.Code) defer func(start time.Time) { sampler := func() metrics.Sample { diff --git a/metrics/metrics.go b/metrics/metrics.go index bf2e26813a..d2de237199 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -6,52 +6,34 @@ package metrics import ( - "os" + "fmt" "runtime/metrics" "runtime/pprof" - "strconv" - "strings" - "syscall" "time" +) - "github.com/ethereum/go-ethereum/log" +var ( + metricsEnabled = false + initRan = false ) // Enabled is checked by functions that are deemed 'expensive', e.g. if a // meter-type does locking and/or non-trivial math operations during update. -// +func Enabled() bool { + return metricsEnabled +} + +// Init enables the metrics system. // The Enabled-flag is expected to be set, once, during startup, but toggling off and on // is not supported: YMMV. -var Enabled = false - -// enablerFlags is the CLI flag names to use to enable metrics collections. -var enablerFlags = []string{"metrics"} - -// enablerEnvVars is the env var names to use to enable metrics collections. -var enablerEnvVars = []string{"GETH_METRICS"} - -// init enables or disables the metrics system. Since we need this to run before -// any other code gets to create meters and timers, we'll actually do an ugly hack -// and peek into the command line args for the metrics flag. -func init() { - for _, enabler := range enablerEnvVars { - if val, found := syscall.Getenv(enabler); found && !Enabled { - if enable, _ := strconv.ParseBool(val); enable { // ignore error, flag parser will choke on it later - log.Info("Enabling metrics collection") - Enabled = true - } - } - } - for _, arg := range os.Args { - flag := strings.TrimLeft(arg, "-") - - for _, enabler := range enablerFlags { - if !Enabled && flag == enabler { - log.Info("Enabling metrics collection") - Enabled = true - } - } +// Init is not safe to call concurrently. It has no effect if it was already called. +func Init(enabled bool) { + metricsEnabled = enabled + if initRan { + return } + initRan = true + // TODO: Maybe start the ticker for exp delays, and things like that. } var threadCreateProfile = pprof.Lookup("threadcreate") @@ -128,7 +110,7 @@ func readRuntimeStats(v *runtimeStats) { // CollectProcessMetrics periodically collects various metrics about the running process. func CollectProcessMetrics(refresh time.Duration) { // Short circuit if the metrics system is disabled - if !Enabled { + if !metricsEnabled { return } diff --git a/metrics/resetting_timer.go b/metrics/resetting_timer.go index 6802e3fcea..c4844bf3e1 100644 --- a/metrics/resetting_timer.go +++ b/metrics/resetting_timer.go @@ -45,7 +45,7 @@ func NewRegisteredResettingTimer(name string, r Registry) ResettingTimer { // NewResettingTimer constructs a new StandardResettingTimer func NewResettingTimer() ResettingTimer { - if !Enabled { + if !metricsEnabled { return NilResettingTimer{} } return &StandardResettingTimer{ diff --git a/p2p/discover/metrics.go b/p2p/discover/metrics.go index 57eb2a8ba7..5d4c953c90 100644 --- a/p2p/discover/metrics.go +++ b/p2p/discover/metrics.go @@ -53,7 +53,7 @@ type meteredUdpConn struct { func newMeteredConn(conn UDPConn) UDPConn { // Short circuit if metrics are disabled - if !metrics.Enabled { + if !metrics.Enabled() { return conn } return &meteredUdpConn{udpConn: conn} diff --git a/p2p/discover/table.go b/p2p/discover/table.go index cbe2972541..392279f905 100644 --- a/p2p/discover/table.go +++ b/p2p/discover/table.go @@ -570,7 +570,7 @@ func (tab *Table) nodeAdded(b *bucket, n *tableNode) { if tab.nodeAddedHook != nil { tab.nodeAddedHook(b, n) } - if metrics.Enabled { + if metrics.Enabled() { bucketsCounter[b.index].Inc(1) } } @@ -580,7 +580,7 @@ func (tab *Table) nodeRemoved(b *bucket, n *tableNode) { if tab.nodeRemovedHook != nil { tab.nodeRemovedHook(b, n) } - if metrics.Enabled { + if metrics.Enabled() { bucketsCounter[b.index].Dec(1) } } diff --git a/p2p/metrics.go b/p2p/metrics.go index 52d4f29eb4..1fd0f26db3 100644 --- a/p2p/metrics.go +++ b/p2p/metrics.go @@ -64,7 +64,7 @@ var ( // markDialError matches errors that occur while setting up a dial connection // to the corresponding meter. func markDialError(err error) { - if !metrics.Enabled { + if !metrics.Enabled() { return } if err2 := errors.Unwrap(err); err2 != nil { @@ -98,7 +98,7 @@ type meteredConn struct { // connection meter and also increases the metered peer count. If the metrics // system is disabled, function returns the original connection. func newMeteredConn(conn net.Conn) net.Conn { - if !metrics.Enabled { + if !metrics.Enabled() { return conn } return &meteredConn{Conn: conn} diff --git a/p2p/peer.go b/p2p/peer.go index c3834965cc..30be151a2f 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -357,7 +357,7 @@ func (p *Peer) handle(msg Msg) error { if err != nil { return fmt.Errorf("msg code out of range: %v", msg.Code) } - if metrics.Enabled { + if metrics.Enabled() { m := fmt.Sprintf("%s/%s/%d/%#02x", ingressMeterName, proto.Name, proto.Version, msg.Code-proto.offset) metrics.GetOrRegisterMeter(m, nil).Mark(int64(msg.meterSize)) metrics.GetOrRegisterMeter(m+"/packets", nil).Mark(1) diff --git a/p2p/tracker/tracker.go b/p2p/tracker/tracker.go index 6a733b9ba5..5b72eb2b88 100644 --- a/p2p/tracker/tracker.go +++ b/p2p/tracker/tracker.go @@ -84,7 +84,7 @@ func New(protocol string, timeout time.Duration) *Tracker { // Track adds a network request to the tracker to wait for a response to arrive // or until the request it cancelled or times out. func (t *Tracker) Track(peer string, version uint, reqCode uint64, resCode uint64, id uint64) { - if !metrics.Enabled { + if !metrics.Enabled() { return } t.lock.Lock() @@ -163,7 +163,7 @@ func (t *Tracker) schedule() { // Fulfil fills a pending request, if any is available, reporting on various metrics. func (t *Tracker) Fulfil(peer string, version uint, code uint64, id uint64) { - if !metrics.Enabled { + if !metrics.Enabled() { return } t.lock.Lock() diff --git a/p2p/transport.go b/p2p/transport.go index 5fc7686feb..360e73a0de 100644 --- a/p2p/transport.go +++ b/p2p/transport.go @@ -98,7 +98,7 @@ func (t *rlpxTransport) WriteMsg(msg Msg) error { // Set metrics. msg.meterSize = size - if metrics.Enabled && msg.meterCap.Name != "" { // don't meter non-subprotocol messages + if metrics.Enabled() && msg.meterCap.Name != "" { // don't meter non-subprotocol messages m := fmt.Sprintf("%s/%s/%d/%#02x", egressMeterName, msg.meterCap.Name, msg.meterCap.Version, msg.meterCode) metrics.GetOrRegisterMeter(m, nil).Mark(int64(msg.meterSize)) metrics.GetOrRegisterMeter(m+"/packets", nil).Mark(1)