cmd, metrics: add support for influxdb-v2 (cherry-picking from italoacasas'...

cmd, metrics: add support for influxdb-v2 (cherry-picking from italoacasas' changes), leave existing support for v1 to maintain backwards-compatibility. (#23194)

This PR adds flag to enable InfluxDB v2 (--metrics.influxdbv2), flags for v2-specific features (--metrics.influxdb.token, --metrics.influxdb.bucket), also carries over addition of support for specifying organization (--metrics.influxdb.organization), but still retains backwards compatibility with InfluxDB v1.

cmd/geth/chaincmd.go

@@ -92,11 +92,15 @@ The dumpgenesis command dumps the genesis block configuration in JSON format to
 			utils.MetricsHTTPFlag,
 			utils.MetricsPortFlag,
 			utils.MetricsEnableInfluxDBFlag,
+			utils.MetricsEnableInfluxDBV2Flag,
 			utils.MetricsInfluxDBEndpointFlag,
 			utils.MetricsInfluxDBDatabaseFlag,
 			utils.MetricsInfluxDBUsernameFlag,
 			utils.MetricsInfluxDBPasswordFlag,
 			utils.MetricsInfluxDBTagsFlag,
+			utils.MetricsInfluxDBTokenFlag,
+			utils.MetricsInfluxDBBucketFlag,
+			utils.MetricsInfluxDBOrganizationFlag,
 			utils.TxLookupLimitFlag,
 		},
 		Category: "BLOCKCHAIN COMMANDS",

cmd/geth/config.go

@@ -233,6 +233,18 @@ func applyMetricConfig(ctx *cli.Context, cfg *gethConfig) {
 	if ctx.GlobalIsSet(utils.MetricsInfluxDBTagsFlag.Name) {
 		cfg.Metrics.InfluxDBTags = ctx.GlobalString(utils.MetricsInfluxDBTagsFlag.Name)
 	}
+	if ctx.GlobalIsSet(utils.MetricsEnableInfluxDBV2Flag.Name) {
+		cfg.Metrics.EnableInfluxDBV2 = ctx.GlobalBool(utils.MetricsEnableInfluxDBV2Flag.Name)
+	}
+	if ctx.GlobalIsSet(utils.MetricsInfluxDBTokenFlag.Name) {
+		cfg.Metrics.InfluxDBToken = ctx.GlobalString(utils.MetricsInfluxDBTokenFlag.Name)
+	}
+	if ctx.GlobalIsSet(utils.MetricsInfluxDBBucketFlag.Name) {
+		cfg.Metrics.InfluxDBBucket = ctx.GlobalString(utils.MetricsInfluxDBBucketFlag.Name)
+	}
+	if ctx.GlobalIsSet(utils.MetricsInfluxDBOrganizationFlag.Name) {
+		cfg.Metrics.InfluxDBOrganization = ctx.GlobalString(utils.MetricsInfluxDBOrganizationFlag.Name)
+	}
 }
 
 func deprecated(field string) bool {

cmd/geth/main.go

@@ -195,6 +195,10 @@ var (
 		utils.MetricsInfluxDBUsernameFlag,
 		utils.MetricsInfluxDBPasswordFlag,
 		utils.MetricsInfluxDBTagsFlag,
+		utils.MetricsEnableInfluxDBV2Flag,
+		utils.MetricsInfluxDBTokenFlag,
+		utils.MetricsInfluxDBBucketFlag,
+		utils.MetricsInfluxDBOrganizationFlag,
 	}
 )
 

cmd/utils/flags.go

@@ -756,6 +756,29 @@ var (
 		Value: metrics.DefaultConfig.InfluxDBTags,
 	}
 
+	MetricsEnableInfluxDBV2Flag = cli.BoolFlag{
+		Name:  "metrics.influxdbv2",
+		Usage: "Enable metrics export/push to an external InfluxDB v2 database",
+	}
+
+	MetricsInfluxDBTokenFlag = cli.StringFlag{
+		Name:  "metrics.influxdb.token",
+		Usage: "Token to authorize access to the database (v2 only)",
+		Value: metrics.DefaultConfig.InfluxDBToken,
+	}
+
+	MetricsInfluxDBBucketFlag = cli.StringFlag{
+		Name:  "metrics.influxdb.bucket",
+		Usage: "InfluxDB bucket name to push reported metrics to (v2 only)",
+		Value: metrics.DefaultConfig.InfluxDBBucket,
+	}
+
+	MetricsInfluxDBOrganizationFlag = cli.StringFlag{
+		Name:  "metrics.influxdb.organization",
+		Usage: "InfluxDB organization name (v2 only)",
+		Value: metrics.DefaultConfig.InfluxDBOrganization,
+	}
+
 	CatalystFlag = cli.BoolFlag{
 		Name:  "catalyst",
 		Usage: "Catalyst mode (eth2 integration testing)",
@@ -1739,11 +1762,36 @@ func SetupMetrics(ctx *cli.Context) {
 		log.Info("Enabling metrics collection")
 
 		var (
-			enableExport = ctx.GlobalBool(MetricsEnableInfluxDBFlag.Name)
-			endpoint     = ctx.GlobalString(MetricsInfluxDBEndpointFlag.Name)
-			database     = ctx.GlobalString(MetricsInfluxDBDatabaseFlag.Name)
-			username     = ctx.GlobalString(MetricsInfluxDBUsernameFlag.Name)
-			password     = ctx.GlobalString(MetricsInfluxDBPasswordFlag.Name)
+			enableExport   = ctx.GlobalBool(MetricsEnableInfluxDBFlag.Name)
+			enableExportV2 = ctx.GlobalBool(MetricsEnableInfluxDBV2Flag.Name)
+		)
+
+		if enableExport || enableExportV2 {
+			CheckExclusive(ctx, MetricsEnableInfluxDBFlag, MetricsEnableInfluxDBV2Flag)
+
+			v1FlagIsSet := ctx.GlobalIsSet(MetricsInfluxDBUsernameFlag.Name) ||
+				ctx.GlobalIsSet(MetricsInfluxDBPasswordFlag.Name)
+
+			v2FlagIsSet := ctx.GlobalIsSet(MetricsInfluxDBTokenFlag.Name) ||
+				ctx.GlobalIsSet(MetricsInfluxDBOrganizationFlag.Name) ||
+				ctx.GlobalIsSet(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")
+			}
+		}
+
+		var (
+			endpoint = ctx.GlobalString(MetricsInfluxDBEndpointFlag.Name)
+			database = ctx.GlobalString(MetricsInfluxDBDatabaseFlag.Name)
+			username = ctx.GlobalString(MetricsInfluxDBUsernameFlag.Name)
+			password = ctx.GlobalString(MetricsInfluxDBPasswordFlag.Name)
+
+			token        = ctx.GlobalString(MetricsInfluxDBTokenFlag.Name)
+			bucket       = ctx.GlobalString(MetricsInfluxDBBucketFlag.Name)
+			organization = ctx.GlobalString(MetricsInfluxDBOrganizationFlag.Name)
 		)
 
 		if enableExport {
@@ -1752,6 +1800,12 @@ func SetupMetrics(ctx *cli.Context) {
 			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.GlobalString(MetricsInfluxDBTagsFlag.Name))
+
+			log.Info("Enabling metrics export to InfluxDB (v2)")
+
+			go influxdb.InfluxDBV2WithTags(metrics.DefaultRegistry, 10*time.Second, endpoint, token, bucket, organization, "geth.", tagsMap)
 		}
 
 		if ctx.GlobalIsSet(MetricsHTTPFlag.Name) {

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/consensys/gnark-crypto v0.4.1-0.20210426202927-39ac3d4b3f1f
 	github.com/davecgh/go-spew v1.1.1
 	github.com/deckarep/golang-set v0.0.0-20180603214616-504e848d77ea
+	github.com/deepmap/oapi-codegen v1.8.2 // indirect
 	github.com/dlclark/regexp2 v1.2.0 // indirect
 	github.com/docker/docker v1.4.2-0.20180625184442-8e610b2b55bf
 	github.com/dop251/goja v0.0.0-20200721192441-a695b0cdd498
@@ -39,13 +40,15 @@ require (
 	github.com/holiman/uint256 v1.2.0
 	github.com/huin/goupnp v1.0.2
 	github.com/influxdata/influxdb v1.8.3
+	github.com/influxdata/influxdb-client-go/v2 v2.4.0
+	github.com/influxdata/line-protocol v0.0.0-20210311194329-9aa0e372d097 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2-0.20160603034137-1fa385a6f458
 	github.com/jedisct1/go-minisign v0.0.0-20190909160543-45766022959e
 	github.com/julienschmidt/httprouter v1.2.0
 	github.com/karalabe/usb v0.0.0-20190919080040-51dc0efba356
 	github.com/kylelemons/godebug v1.1.0 // indirect
-	github.com/mattn/go-colorable v0.1.0
-	github.com/mattn/go-isatty v0.0.5-0.20180830101745-3fb116b82035
+	github.com/mattn/go-colorable v0.1.8
+	github.com/mattn/go-isatty v0.0.12
 	github.com/naoina/go-stringutil v0.1.0 // indirect
 	github.com/naoina/toml v0.1.2-0.20170918210437-9fafd6967416
 	github.com/olekukonko/tablewriter v0.0.5
@@ -60,12 +63,14 @@ require (
 	github.com/tklauser/go-sysconf v0.3.5 // indirect
 	github.com/tyler-smith/go-bip39 v1.0.1-0.20181017060643-dbb3b84ba2ef
 	golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2
+	golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	golang.org/x/sys v0.0.0-20210420205809-ac73e9fd8988
+	golang.org/x/sys v0.0.0-20210423082822-04245dca01da
 	golang.org/x/text v0.3.6
-	golang.org/x/time v0.0.0-20201208040808-7e3f01d25324
+	golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce
 	gopkg.in/olebedev/go-duktape.v3 v3.0.0-20200619000410-60c24ae608a6
 	gopkg.in/urfave/cli.v1 v1.20.0
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gotest.tools v2.2.0+incompatible // indirect
 )

metrics/config.go

@@ -28,6 +28,11 @@ type Config struct {
 	InfluxDBUsername string `toml:",omitempty"`
 	InfluxDBPassword string `toml:",omitempty"`
 	InfluxDBTags     string `toml:",omitempty"`
+
+	EnableInfluxDBV2     bool   `toml:",omitempty"`
+	InfluxDBToken        string `toml:",omitempty"`
+	InfluxDBBucket       string `toml:",omitempty"`
+	InfluxDBOrganization string `toml:",omitempty"`
 }
 
 // DefaultConfig is the default config for metrics used in go-ethereum.
@@ -42,4 +47,10 @@ var DefaultConfig = Config{
 	InfluxDBUsername: "test",
 	InfluxDBPassword: "test",
 	InfluxDBTags:     "host=localhost",
+
+	// influxdbv2-specific flags
+	EnableInfluxDBV2:     false,
+	InfluxDBToken:        "test",
+	InfluxDBBucket:       "geth",
+	InfluxDBOrganization: "geth",
 }

metrics/influxdb/influxdbv2.go

+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+package influxdb
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/metrics"
+	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
+	"github.com/influxdata/influxdb-client-go/v2/api"
+)
+
+type v2Reporter struct {
+	reg      metrics.Registry
+	interval time.Duration
+
+	endpoint     string
+	token        string
+	bucket       string
+	organization string
+	namespace    string
+	tags         map[string]string
+
+	client influxdb2.Client
+	write  api.WriteAPI
+
+	cache map[string]int64
+}
+
+// InfluxDBWithTags starts a InfluxDB reporter which will post the from the given metrics.Registry at each d interval with the specified tags
+func InfluxDBV2WithTags(r metrics.Registry, d time.Duration, endpoint string, token string, bucket string, organization string, namespace string, tags map[string]string) {
+	rep := &v2Reporter{
+		reg:          r,
+		interval:     d,
+		endpoint:     endpoint,
+		token:        token,
+		bucket:       bucket,
+		organization: organization,
+		namespace:    namespace,
+		tags:         tags,
+		cache:        make(map[string]int64),
+	}
+
+	rep.client = influxdb2.NewClient(rep.endpoint, rep.token)
+	defer rep.client.Close()
+
+	// async write client
+	rep.write = rep.client.WriteAPI(rep.organization, rep.bucket)
+	errorsCh := rep.write.Errors()
+
+	// have to handle write errors in a separate goroutine like this b/c the channel is unbuffered and will block writes if not read
+	go func() {
+		for err := range errorsCh {
+			log.Warn("write error", "err", err.Error())
+		}
+	}()
+	rep.run()
+}
+
+func (r *v2Reporter) run() {
+	intervalTicker := time.Tick(r.interval)
+	pingTicker := time.Tick(time.Second * 5)
+
+	for {
+		select {
+		case <-intervalTicker:
+			r.send()
+		case <-pingTicker:
+			_, err := r.client.Health(context.Background())
+			if err != nil {
+				log.Warn("Got error from influxdb client health check", "err", err.Error())
+			}
+		}
+	}
+
+}
+
+func (r *v2Reporter) send() {
+	r.reg.Each(func(name string, i interface{}) {
+		now := time.Now()
+		namespace := r.namespace
+
+		switch metric := i.(type) {
+
+		case metrics.Counter:
+			v := metric.Count()
+			l := r.cache[name]
+
+			measurement := fmt.Sprintf("%s%s.count", namespace, name)
+			fields := map[string]interface{}{
+				"value": v - l,
+			}
+
+			pt := influxdb2.NewPoint(measurement, r.tags, fields, now)
+			r.write.WritePoint(pt)
+
+			r.cache[name] = v
+
+		case metrics.Gauge:
+			ms := metric.Snapshot()
+
+			measurement := fmt.Sprintf("%s%s.gauge", namespace, name)
+			fields := map[string]interface{}{
+				"value": ms.Value(),
+			}
+
+			pt := influxdb2.NewPoint(measurement, r.tags, fields, now)
+			r.write.WritePoint(pt)
+
+		case metrics.GaugeFloat64:
+			ms := metric.Snapshot()
+
+			measurement := fmt.Sprintf("%s%s.gauge", namespace, name)
+			fields := map[string]interface{}{
+				"value": ms.Value(),
+			}
+
+			pt := influxdb2.NewPoint(measurement, r.tags, fields, now)
+			r.write.WritePoint(pt)
+
+		case metrics.Histogram:
+			ms := metric.Snapshot()
+
+			if ms.Count() > 0 {
+				ps := ms.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999, 0.9999})
+				measurement := fmt.Sprintf("%s%s.histogram", namespace, name)
+				fields := map[string]interface{}{
+					"count":    ms.Count(),
+					"max":      ms.Max(),
+					"mean":     ms.Mean(),
+					"min":      ms.Min(),
+					"stddev":   ms.StdDev(),
+					"variance": ms.Variance(),
+					"p50":      ps[0],
+					"p75":      ps[1],
+					"p95":      ps[2],
+					"p99":      ps[3],
+					"p999":     ps[4],
+					"p9999":    ps[5],
+				}
+
+				pt := influxdb2.NewPoint(measurement, r.tags, fields, now)
+				r.write.WritePoint(pt)
+			}
+
+		case metrics.Meter:
+			ms := metric.Snapshot()
+
+			measurement := fmt.Sprintf("%s%s.meter", namespace, name)
+			fields := map[string]interface{}{
+				"count": ms.Count(),
+				"m1":    ms.Rate1(),
+				"m5":    ms.Rate5(),
+				"m15":   ms.Rate15(),
+				"mean":  ms.RateMean(),
+			}
+
+			pt := influxdb2.NewPoint(measurement, r.tags, fields, now)
+			r.write.WritePoint(pt)
+
+		case metrics.Timer:
+			ms := metric.Snapshot()
+			ps := ms.Percentiles([]float64{0.5, 0.75, 0.95, 0.99, 0.999, 0.9999})
+
+			measurement := fmt.Sprintf("%s%s.timer", namespace, name)
+			fields := map[string]interface{}{
+				"count":    ms.Count(),
+				"max":      ms.Max(),
+				"mean":     ms.Mean(),
+				"min":      ms.Min(),
+				"stddev":   ms.StdDev(),
+				"variance": ms.Variance(),
+				"p50":      ps[0],
+				"p75":      ps[1],
+				"p95":      ps[2],
+				"p99":      ps[3],
+				"p999":     ps[4],
+				"p9999":    ps[5],
+				"m1":       ms.Rate1(),
+				"m5":       ms.Rate5(),
+				"m15":      ms.Rate15(),
+				"meanrate": ms.RateMean(),
+			}
+
+			pt := influxdb2.NewPoint(measurement, r.tags, fields, now)
+			r.write.WritePoint(pt)
+
+		case metrics.ResettingTimer:
+			t := metric.Snapshot()
+
+			if len(t.Values()) > 0 {
+				ps := t.Percentiles([]float64{50, 95, 99})
+				val := t.Values()
+
+				measurement := fmt.Sprintf("%s%s.span", namespace, name)
+				fields := map[string]interface{}{
+					"count": len(val),
+					"max":   val[len(val)-1],
+					"mean":  t.Mean(),
+					"min":   val[0],
+					"p50":   ps[0],
+					"p95":   ps[1],
+					"p99":   ps[2],
+				}
+
+				pt := influxdb2.NewPoint(measurement, r.tags, fields, now)
+				r.write.WritePoint(pt)
+			}
+		}
+	})
+
+	// Force all unwritten data to be sent
+	r.write.Flush()
+}