cmd/geth: add flags for metrics export

cmd/geth/main.go

@@ -140,6 +140,15 @@ var (
 		utils.WhisperMaxMessageSizeFlag,
 		utils.WhisperMinPOWFlag,
 	}
+
+	metricsFlags = []cli.Flag{
+		utils.MetricsEnableInfluxDBExportFlag,
+		utils.MetricsInfluxDBEndpointFlag,
+		utils.MetricsInfluxDBDatabaseFlag,
+		utils.MetricsInfluxDBUsernameFlag,
+		utils.MetricsInfluxDBPasswordFlag,
+		utils.MetricsInfluxDBHostTagFlag,
+	}
 )
 
 func init() {
@@ -182,12 +191,17 @@ func init() {
 	app.Flags = append(app.Flags, consoleFlags...)
 	app.Flags = append(app.Flags, debug.Flags...)
 	app.Flags = append(app.Flags, whisperFlags...)
+	app.Flags = append(app.Flags, metricsFlags...)
 
 	app.Before = func(ctx *cli.Context) error {
 		runtime.GOMAXPROCS(runtime.NumCPU())
 		if err := debug.Setup(ctx); err != nil {
 			return err
 		}
+
+		// Start metrics export if enabled
+		utils.SetupMetrics(ctx)
+
 		// Start system runtime metrics collection
 		go metrics.CollectProcessMetrics(3 * time.Second)
 

cmd/utils/flags.go

@@ -27,6 +27,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/accounts/keystore"
@@ -48,6 +49,7 @@ import (
 	"github.com/ethereum/go-ethereum/les"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/metrics/influxdb"
 	"github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/discover"
@@ -532,6 +534,41 @@ var (
 		Usage: "Minimum POW accepted",
 		Value: whisper.DefaultMinimumPoW,
 	}
+
+	// Metrics flags
+	MetricsEnableInfluxDBExportFlag = cli.BoolFlag{
+		Name:  "metrics.influxdb.export",
+		Usage: "Enable metrics export/push to an external InfluxDB database",
+	}
+	MetricsInfluxDBEndpointFlag = cli.StringFlag{
+		Name:  "metrics.influxdb.endpoint",
+		Usage: "Metrics InfluxDB endpoint",
+		Value: "http://127.0.0.1:8086",
+	}
+	MetricsInfluxDBDatabaseFlag = cli.StringFlag{
+		Name:  "metrics.influxdb.database",
+		Usage: "Metrics InfluxDB database",
+		Value: "metrics",
+	}
+	MetricsInfluxDBUsernameFlag = cli.StringFlag{
+		Name:  "metrics.influxdb.username",
+		Usage: "Metrics InfluxDB username",
+		Value: "",
+	}
+	MetricsInfluxDBPasswordFlag = cli.StringFlag{
+		Name:  "metrics.influxdb.password",
+		Usage: "Metrics InfluxDB password",
+		Value: "",
+	}
+	// 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: "Metrics InfluxDB `host` tag attached to all measurements",
+		Value: "localhost",
+	}
 )
 
 // MakeDataDir retrieves the currently requested data directory, terminating
@@ -1145,7 +1182,14 @@ func RegisterEthService(stack *node.Node, cfg *eth.Config) {
 // RegisterDashboardService adds a dashboard to the stack.
 func RegisterDashboardService(stack *node.Node, cfg *dashboard.Config, commit string) {
 	stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
-		return dashboard.New(cfg, commit)
+		// Retrieve both eth and les services
+		var ethServ *eth.Ethereum
+		ctx.Service(&ethServ)
+
+		var lesServ *les.LightEthereum
+		ctx.Service(&lesServ)
+
+		return dashboard.New(cfg, commit, ethServ, lesServ)
 	})
 }
 
@@ -1181,6 +1225,27 @@ func SetupNetwork(ctx *cli.Context) {
 	params.TargetGasLimit = ctx.GlobalUint64(TargetGasLimitFlag.Name)
 }
 
+func SetupMetrics(ctx *cli.Context) {
+	if metrics.Enabled {
+		log.Info("Enabling metrics collection")
+		var (
+			enableExport = ctx.GlobalBool(MetricsEnableInfluxDBExportFlag.Name)
+			endpoint     = ctx.GlobalString(MetricsInfluxDBEndpointFlag.Name)
+			database     = ctx.GlobalString(MetricsInfluxDBDatabaseFlag.Name)
+			username     = ctx.GlobalString(MetricsInfluxDBUsernameFlag.Name)
+			password     = ctx.GlobalString(MetricsInfluxDBPasswordFlag.Name)
+			hosttag      = ctx.GlobalString(MetricsInfluxDBHostTagFlag.Name)
+		)
+
+		if enableExport {
+			log.Info("Enabling metrics export to InfluxDB")
+			go influxdb.InfluxDBWithTags(metrics.DefaultRegistry, 10*time.Second, endpoint, database, username, password, "geth.", map[string]string{
+				"host": hosttag,
+			})
+		}
+	}
+}
+
 // 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 (

dashboard/dashboard.go

@@ -33,6 +33,8 @@ import (
 	"time"
 
 	"github.com/elastic/gosigar"
+	"github.com/ethereum/go-ethereum/eth"
+	"github.com/ethereum/go-ethereum/les"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/p2p"
@@ -64,6 +66,9 @@ type Dashboard struct {
 	commit   string
 	lock     sync.RWMutex // Lock protecting the dashboard's internals
 
+	ethServ *eth.Ethereum
+	lesServ *les.LightEthereum
+
 	quit chan chan error // Channel used for graceful exit
 	wg   sync.WaitGroup
 }
@@ -76,7 +81,7 @@ type client struct {
 }
 
 // New creates a new dashboard instance with the given configuration.
-func New(config *Config, commit string) (*Dashboard, error) {
+func New(config *Config, commit string, _ethServ *eth.Ethereum, _lesserv *les.LightEthereum) (*Dashboard, error) {
 	now := time.Now()
 	db := &Dashboard{
 		conns:  make(map[uint32]*client),
@@ -93,6 +98,8 @@ func New(config *Config, commit string) (*Dashboard, error) {
 			DiskWrite:      emptyChartEntries(now, diskWriteSampleLimit, config.Refresh),
 		},
 		commit:  commit,
+		ethServ: _ethServ,
+		lesServ: _lesserv,
 	}
 	return db, nil
 }
@@ -306,6 +313,23 @@ func (db *Dashboard) collectData() {
 			prevDiskRead = curDiskRead
 			prevDiskWrite = curDiskWrite
 
+			// extract metrics from downloaded and push to registry
+			p := db.ethServ.Downloader().Progress()
+
+			metrics.GetOrRegisterGauge("currentBlock", nil).Update(int64(p.CurrentBlock))
+			metrics.GetOrRegisterGauge("startingBlock", nil).Update(int64(p.StartingBlock))
+			metrics.GetOrRegisterGauge("highestBlock", nil).Update(int64(p.HighestBlock))
+			metrics.GetOrRegisterGauge("pulledStates", nil).Update(int64(p.PulledStates))
+			metrics.GetOrRegisterGauge("knownStates", nil).Update(int64(p.KnownStates))
+
+			syncing := db.ethServ.BlockChain().CurrentHeader().Number.Uint64() >= p.HighestBlock
+
+			if syncing {
+				metrics.GetOrRegisterGauge("isSyncing", nil).Update(1)
+			} else {
+				metrics.GetOrRegisterGauge("isSyncing", nil).Update(0)
+			}
+
 			now := time.Now()
 
 			runtime.ReadMemStats(&mem)