metrics: export to InfluxDB (ethereum#16979)

gzliudan · Dec 10, 2024 · b3e5f4e · b3e5f4e
1 parent b0fefe4
commit b3e5f4e
Show file tree

Hide file tree

Showing 4 changed files with 102 additions and 21 deletions.
diff --git a/cmd/XDC/main.go b/cmd/XDC/main.go
@@ -126,9 +126,6 @@ var (
 		utils.HTTPCORSDomainFlag,
 		utils.HTTPVirtualHostsFlag,
 		utils.EthStatsURLFlag,
-		utils.MetricsEnabledFlag,
-		utils.MetricsHTTPFlag,
-		utils.MetricsPortFlag,
 		//utils.FakePoWFlag,
 		//utils.NoCompactionFlag,
 		//utils.GpoBlocksFlag,
@@ -163,6 +160,18 @@ var (
 		utils.IPCPathFlag,
 		utils.RPCGlobalTxFeeCap,
 	}
+
+	metricsFlags = []cli.Flag{
+		utils.MetricsEnabledFlag,
+		utils.MetricsHTTPFlag,
+		utils.MetricsPortFlag,
+		utils.MetricsEnableInfluxDBFlag,
+		utils.MetricsInfluxDBEndpointFlag,
+		utils.MetricsInfluxDBDatabaseFlag,
+		utils.MetricsInfluxDBUsernameFlag,
+		utils.MetricsInfluxDBPasswordFlag,
+		utils.MetricsInfluxDBHostTagFlag,
+	}
 )
 
 func init() {
@@ -194,6 +203,7 @@ func init() {
 	app.Flags = append(app.Flags, rpcFlags...)
 	app.Flags = append(app.Flags, consoleFlags...)
 	app.Flags = append(app.Flags, debug.Flags...)
+	app.Flags = append(app.Flags, metricsFlags...)
 	flags.AutoEnvVars(app.Flags, "XDC")
 
 	app.Before = func(ctx *cli.Context) error {
@@ -203,6 +213,7 @@ func init() {
 			return err
 		}
 		flags.CheckEnvVars(ctx, app.Flags, "XDC")
+
 		// Start system runtime metrics collection
 		go metrics.CollectProcessMetrics(3 * time.Second)
 

diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go
@@ -28,6 +28,7 @@ import (
 	godebug "runtime/debug"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/XinFinOrg/XDPoSChain/XDCx"
 	"github.com/XinFinOrg/XDPoSChain/accounts"
@@ -52,6 +53,7 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/log"
 	"github.com/XinFinOrg/XDPoSChain/metrics"
 	"github.com/XinFinOrg/XDPoSChain/metrics/exp"
+	"github.com/XinFinOrg/XDPoSChain/metrics/influxdb"
 	"github.com/XinFinOrg/XDPoSChain/node"
 	"github.com/XinFinOrg/XDPoSChain/p2p"
 	"github.com/XinFinOrg/XDPoSChain/p2p/discover"
@@ -661,6 +663,45 @@ var (
 		Value:    metrics.DefaultConfig.Port,
 		Category: flags.MetricsCategory,
 	}
+	MetricsEnableInfluxDBFlag = &cli.BoolFlag{
+		Name:     "metrics-influxdb",
+		Usage:    "Enable metrics export/push to an external InfluxDB database",
+		Category: flags.MetricsCategory,
+	}
+	MetricsInfluxDBEndpointFlag = &cli.StringFlag{
+		Name:     "metrics-influxdb.endpoint",
+		Usage:    "InfluxDB API endpoint to report metrics to",
+		Value:    metrics.DefaultConfig.InfluxDBEndpoint,
+		Category: flags.MetricsCategory,
+	}
+	MetricsInfluxDBDatabaseFlag = &cli.StringFlag{
+		Name:     "metrics-influxdb.database",
+		Usage:    "InfluxDB database name to push reported metrics to",
+		Value:    metrics.DefaultConfig.InfluxDBDatabase,
+		Category: flags.MetricsCategory,
+	}
+	MetricsInfluxDBUsernameFlag = &cli.StringFlag{
+		Name:     "metrics-influxdb.username",
+		Usage:    "Username to authorize access to the database",
+		Value:    metrics.DefaultConfig.InfluxDBUsername,
+		Category: flags.MetricsCategory,
+	}
+	MetricsInfluxDBPasswordFlag = &cli.StringFlag{
+		Name:     "metrics-influxdb.password",
+		Usage:    "Password to authorize access to the database",
+		Value:    metrics.DefaultConfig.InfluxDBPassword,
+		Category: flags.MetricsCategory,
+	}
+	// The `host` tag is part of every measurement sent to InfluxDB. Queries on tags are faster in InfluxDB.
+	// It is used so that we can group all nodes and average a measurement across all of them, but also so
+	// that we can select a specific node and inspect its measurements.
+	// https://docs.influxdata.com/influxdb/v1.4/concepts/key_concepts/#tag-key
+	MetricsInfluxDBHostTagFlag = &cli.StringFlag{
+		Name:     "metrics-influxdb.host.tag",
+		Usage:    "InfluxDB `host` tag attached to all measurements",
+		Value:    metrics.DefaultConfig.InfluxDBTags,
+		Category: flags.MetricsCategory,
+	}
 
 	// MISC settings
 	RollbackFlag = &cli.StringFlag{
@@ -1468,6 +1509,35 @@ func SetupNetwork(ctx *cli.Context) {
 	params.TargetGasLimit = ctx.Uint64(MinerGasLimitFlag.Name)
 }
 
+func SetupMetrics(ctx *cli.Context) {
+	if metrics.Enabled {
+		log.Info("Enabling metrics collection")
+		var (
+			enableExport = ctx.Bool(MetricsEnableInfluxDBFlag.Name)
+			endpoint     = ctx.String(MetricsInfluxDBEndpointFlag.Name)
+			database     = ctx.String(MetricsInfluxDBDatabaseFlag.Name)
+			username     = ctx.String(MetricsInfluxDBUsernameFlag.Name)
+			password     = ctx.String(MetricsInfluxDBPasswordFlag.Name)
+			hosttag      = ctx.String(MetricsInfluxDBHostTagFlag.Name)
+		)
+
+		if enableExport {
+			log.Info("Enabling metrics export to InfluxDB")
+			go influxdb.InfluxDBWithTags(metrics.DefaultRegistry, 10*time.Second, endpoint, database, username, password, "xdc.", map[string]string{
+				"host": hosttag,
+			})
+		}
+
+		if ctx.IsSet(MetricsHTTPFlag.Name) {
+			address := fmt.Sprintf("%s:%d", ctx.String(MetricsHTTPFlag.Name), 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))
+		}
+	}
+}
+
 // MakeChainDatabase open an LevelDB using the flags passed to the client and will hard crash if it fails.
 func MakeChainDatabase(ctx *cli.Context, stack *node.Node) ethdb.Database {
 	var (
@@ -1595,15 +1665,3 @@ func RegisterFilterAPI(stack *node.Node, backend ethapi.Backend, ethcfg *ethconf
 	}})
 	return filterSystem
 }
-
-func SetupMetrics(ctx *cli.Context) {
-	if metrics.Enabled {
-		log.Info("Enabling metrics collection")
-
-		if ctx.IsSet(MetricsHTTPFlag.Name) {
-			address := fmt.Sprintf("%s:%d", ctx.String(MetricsHTTPFlag.Name), ctx.Int(MetricsPortFlag.Name))
-			log.Info("Enabling stand-alone metrics HTTP endpoint", "address", address)
-			exp.Setup(address)
-		}
-	}
-}
diff --git a/metrics/config.go b/metrics/config.go
@@ -22,6 +22,12 @@ type Config struct {
 	EnabledExpensive bool   `toml:",omitempty"`
 	HTTP             string `toml:",omitempty"`
 	Port             int    `toml:",omitempty"`
+	EnableInfluxDB   bool   `toml:",omitempty"`
+	InfluxDBEndpoint string `toml:",omitempty"`
+	InfluxDBDatabase string `toml:",omitempty"`
+	InfluxDBUsername string `toml:",omitempty"`
+	InfluxDBPassword string `toml:",omitempty"`
+	InfluxDBTags     string `toml:",omitempty"`
 }
 
 // DefaultConfig is the default config for metrics used in go-ethereum.
@@ -30,4 +36,10 @@ var DefaultConfig = Config{
 	EnabledExpensive: false,
 	HTTP:             "127.0.0.1",
 	Port:             6060,
+	EnableInfluxDB:   false,
+	InfluxDBEndpoint: "http://localhost:8086",
+	InfluxDBDatabase: "xdc",
+	InfluxDBUsername: "test",
+	InfluxDBPassword: "test",
+	InfluxDBTags:     "localhost",
 }
diff --git a/metrics/influxdb/influxdb.go b/metrics/influxdb/influxdb.go
@@ -2,10 +2,10 @@ package influxdb
 
 import (
 	"fmt"
-	"log"
 	uurl "net/url"
 	"time"
 
+	"github.com/XinFinOrg/XDPoSChain/log"
 	"github.com/XinFinOrg/XDPoSChain/metrics"
 	"github.com/influxdata/influxdb/client"
 )
@@ -35,7 +35,7 @@ func InfluxDB(r metrics.Registry, d time.Duration, url, database, username, pass
 func InfluxDBWithTags(r metrics.Registry, d time.Duration, url, database, username, password, namespace string, tags map[string]string) {
 	u, err := uurl.Parse(url)
 	if err != nil {
-		log.Printf("unable to parse InfluxDB url %s. err=%v", url, err)
+		log.Warn("unable to parse InfluxDB url %s. err=%v", url, err)
 		return
 	}
 
@@ -51,7 +51,7 @@ func InfluxDBWithTags(r metrics.Registry, d time.Duration, url, database, userna
 		cache:     make(map[string]int64),
 	}
 	if err := rep.makeClient(); err != nil {
-		log.Printf("unable to make InfluxDB client. err=%v", err)
+		log.Warn("unable to make InfluxDB client. err=%v", err)
 		return
 	}
 
@@ -79,15 +79,15 @@ func (r *reporter) run() {
 		select {
 		case <-intervalTicker.C:
 			if err := r.send(); err != nil {
-				log.Printf("unable to send to InfluxDB. err=%v", err)
+				log.Warn("unable to send to InfluxDB. err=%v", err)
 			}
 		case <-pingTicker.C:
 			_, _, err := r.client.Ping()
 			if err != nil {
-				log.Printf("got error while sending a ping to InfluxDB, trying to recreate client. err=%v", err)
+				log.Warn("got error while sending a ping to InfluxDB, trying to recreate client. err=%v", err)
 
 				if err = r.makeClient(); err != nil {
-					log.Printf("unable to make InfluxDB client. err=%v", err)
+					log.Warn("unable to make InfluxDB client. err=%v", err)
 				}
 			}
 		}