diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 09d9c493d1..57271e4de2 100644 --- a/cmd/geth/main.go +++ b/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) diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index ef5f6a9f08..e10e1d82d0 100644 --- a/cmd/utils/flags.go +++ b/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(ðServ) + + 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 ( diff --git a/dashboard/dashboard.go b/dashboard/dashboard.go index 399fa34c08..752f23a17c 100644 --- a/dashboard/dashboard.go +++ b/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), @@ -92,7 +97,9 @@ func New(config *Config, commit string) (*Dashboard, error) { DiskRead: emptyChartEntries(now, diskReadSampleLimit, config.Refresh), DiskWrite: emptyChartEntries(now, diskWriteSampleLimit, config.Refresh), }, - commit: commit, + 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)