cmd/geth, rpc/api: extend metrics API, add a basic monitor command

cmd/geth/main.go

@@ -214,6 +214,16 @@ The Geth console is an interactive shell for the JavaScript runtime environment
 which exposes a node admin interface as well as the Ðapp JavaScript API.
 See https://github.com/ethereum/go-ethereum/wiki/Javascipt-Console.
 This command allows to open a console on a running geth node.
+`,
+		},
+		{
+			Action: monitor,
+			Name:   "monitor",
+			Usage:  `Geth Monitor: node metrics monitoring and visualization`,
+			Description: `
+The Geth monitor is a tool to collect and visualize various internal metrics
+gathered by the node, supporting different chart types as well as the capacity
+to display multiple metrics simultaneously.
 `,
 		},
 		{

cmd/geth/monitorcmd.go

+package main
+
+import (
+	"reflect"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/codegangsta/cli"
+	"github.com/ethereum/go-ethereum/cmd/utils"
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/ethereum/go-ethereum/rpc/codec"
+	"github.com/ethereum/go-ethereum/rpc/comms"
+	"github.com/gizak/termui"
+)
+
+// monitor starts a terminal UI based monitoring tool for the requested metrics.
+func monitor(ctx *cli.Context) {
+	var (
+		client comms.EthereumClient
+		args   []string
+		err    error
+	)
+	// Attach to an Ethereum node over IPC or RPC
+	if ctx.Args().Present() {
+		// Try to interpret the first parameter as an endpoint
+		client, err = comms.ClientFromEndpoint(ctx.Args().First(), codec.JSON)
+		if err == nil {
+			args = ctx.Args().Tail()
+		}
+	}
+	if !ctx.Args().Present() || err != nil {
+		// Either no args were given, or not endpoint, use defaults
+		cfg := comms.IpcConfig{
+			Endpoint: ctx.GlobalString(utils.IPCPathFlag.Name),
+		}
+		args = ctx.Args()
+		client, err = comms.NewIpcClient(cfg, codec.JSON)
+	}
+	if err != nil {
+		utils.Fatalf("Unable to attach to geth node - %v", err)
+	}
+	defer client.Close()
+
+	xeth := rpc.NewXeth(client)
+
+	// Retrieve all the available metrics and resolve the user pattens
+	metrics, err := xeth.Call("debug_metrics", []interface{}{true})
+	if err != nil {
+		utils.Fatalf("Failed to retrieve system metrics: %v", err)
+	}
+	monitored := resolveMetrics(metrics, args)
+	sort.Strings(monitored)
+
+	// Create the access function and check that the metric exists
+	value := func(metrics map[string]interface{}, metric string) float64 {
+		parts, found := strings.Split(metric, "/"), true
+		for _, part := range parts[:len(parts)-1] {
+			metrics, found = metrics[part].(map[string]interface{})
+			if !found {
+				utils.Fatalf("Metric not found: %s", metric)
+			}
+		}
+		if v, ok := metrics[parts[len(parts)-1]].(float64); ok {
+			return v
+		}
+		utils.Fatalf("Metric not float64: %s", metric)
+		return 0
+	}
+	// Assemble the terminal UI
+	if err := termui.Init(); err != nil {
+		utils.Fatalf("Unable to initialize terminal UI: %v", err)
+	}
+	defer termui.Close()
+
+	termui.UseTheme("helloworld")
+
+	charts := make([]*termui.LineChart, len(monitored))
+	for i, metric := range monitored {
+		charts[i] = termui.NewLineChart()
+		charts[i].Border.Label = metric
+		charts[i].Data = make([]float64, 512)
+		charts[i].DataLabels = []string{""}
+		charts[i].Height = termui.TermHeight() / len(monitored)
+		charts[i].AxesColor = termui.ColorWhite
+		charts[i].LineColor = termui.ColorGreen
+
+		termui.Body.AddRows(termui.NewRow(termui.NewCol(12, 0, charts[i])))
+	}
+	termui.Body.Align()
+	termui.Render(termui.Body)
+
+	refresh := time.Tick(time.Second)
+	for {
+		select {
+		case event := <-termui.EventCh():
+			if event.Type == termui.EventKey && event.Ch == 'q' {
+				return
+			}
+			if event.Type == termui.EventResize {
+				termui.Body.Width = termui.TermWidth()
+				for _, chart := range charts {
+					chart.Height = termui.TermHeight() / len(monitored)
+				}
+				termui.Body.Align()
+				termui.Render(termui.Body)
+			}
+		case <-refresh:
+			metrics, err := xeth.Call("debug_metrics", []interface{}{true})
+			if err != nil {
+				utils.Fatalf("Failed to retrieve system metrics: %v", err)
+			}
+			for i, metric := range monitored {
+				charts[i].Data = append([]float64{value(metrics, metric)}, charts[i].Data[:len(charts[i].Data)-1]...)
+			}
+			termui.Render(termui.Body)
+		}
+	}
+}
+
+// resolveMetrics takes a list of input metric patterns, and resolves each to one
+// or more canonical metric names.
+func resolveMetrics(metrics map[string]interface{}, patterns []string) []string {
+	res := []string{}
+	for _, pattern := range patterns {
+		res = append(res, resolveMetric(metrics, pattern, "")...)
+	}
+	return res
+}
+
+// resolveMetrics takes a single of input metric pattern, and resolves it to one
+// or more canonical metric names.
+func resolveMetric(metrics map[string]interface{}, pattern string, path string) []string {
+	var ok bool
+
+	// Build up the canonical metric path
+	parts := strings.Split(pattern, "/")
+	for len(parts) > 1 {
+		if metrics, ok = metrics[parts[0]].(map[string]interface{}); !ok {
+			utils.Fatalf("Failed to retrieve system metrics: %s", path+parts[0])
+		}
+		path += parts[0] + "/"
+		parts = parts[1:]
+	}
+	// Depending what the last link is, return or expand
+	switch metric := metrics[parts[0]].(type) {
+	case float64:
+		// Final metric value found, return as singleton
+		return []string{path + parts[0]}
+
+	case map[string]interface{}:
+		return expandMetrics(metric, path+parts[0]+"/")
+
+	default:
+		utils.Fatalf("Metric pattern resolved to unexpected type: %v", reflect.TypeOf(metric))
+		return nil
+	}
+}
+
+// expandMetrics expands the entire tree of metrics into a flat list of paths.
+func expandMetrics(metrics map[string]interface{}, path string) []string {
+	// Iterate over all fields and expand individually
+	list := []string{}
+	for name, metric := range metrics {
+		switch metric := metric.(type) {
+		case float64:
+			// Final metric value found, append to list
+			list = append(list, path+name)
+
+		case map[string]interface{}:
+			// Tree of metrics found, expand recursively
+			list = append(list, expandMetrics(metric, path+name+"/")...)
+
+		default:
+			utils.Fatalf("Metric pattern %s resolved to unexpected type: %v", path+name, reflect.TypeOf(metric))
+			return nil
+		}
+	}
+	return list
+}

rpc/api/debug.go

@@ -177,6 +177,10 @@ func (self *debugApi) SeedHash(req *shared.Request) (interface{}, error) {
 }
 
 func (self *debugApi) Metrics(req *shared.Request) (interface{}, error) {
+	args := new(MetricsArgs)
+	if err := self.codec.Decode(req.Params, &args); err != nil {
+		return nil, shared.NewDecodeParamError(err.Error())
+	}
 	// Create a rate formatter
 	units := []string{"", "K", "M", "G", "T", "E", "P"}
 	round := func(value float64, prec int) string {
@@ -202,7 +206,40 @@ func (self *debugApi) Metrics(req *shared.Request) (interface{}, error) {
 		}
 		name = parts[len(parts)-1]
 
-		// Fill the counter with the metric details
+		// Fill the counter with the metric details, formatting if requested
+		if args.Raw {
+			switch metric := metric.(type) {
+			case metrics.Meter:
+				root[name] = map[string]interface{}{
+					"Avg01Min": metric.Rate1(),
+					"Avg05Min": metric.Rate5(),
+					"Avg15Min": metric.Rate15(),
+					"AvgTotal": metric.RateMean(),
+					"Total":    float64(metric.Count()),
+				}
+
+			case metrics.Timer:
+				root[name] = map[string]interface{}{
+					"Avg01Min": metric.Rate1(),
+					"Avg05Min": metric.Rate5(),
+					"Avg15Min": metric.Rate15(),
+					"AvgTotal": metric.RateMean(),
+					"Total":    float64(metric.Count()),
+					"Maximum":  metric.Max(),
+					"Minimum":  metric.Min(),
+					"Percentile": map[string]interface{}{
+						"20": metric.Percentile(0.2),
+						"50": metric.Percentile(0.5),
+						"80": metric.Percentile(0.8),
+						"95": metric.Percentile(0.95),
+						"99": metric.Percentile(0.99),
+					},
+				}
+
+			default:
+				root[name] = "Unknown metric type"
+			}
+		} else {
 			switch metric := metric.(type) {
 			case metrics.Meter:
 				root[name] = map[string]interface{}{
@@ -232,6 +269,7 @@ func (self *debugApi) Metrics(req *shared.Request) (interface{}, error) {
 			default:
 				root[name] = "Unknown metric type"
 			}
+		}
 	})
 	return counters, nil
 }

rpc/api/debug_args.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/big"
+	"reflect"
 
 	"github.com/ethereum/go-ethereum/rpc/shared"
 )
@@ -45,3 +46,26 @@ func (args *WaitForBlockArgs) UnmarshalJSON(b []byte) (err error) {
 
 	return nil
 }
+
+type MetricsArgs struct {
+	Raw bool
+}
+
+func (args *MetricsArgs) UnmarshalJSON(b []byte) (err error) {
+	var obj []interface{}
+	if err := json.Unmarshal(b, &obj); err != nil {
+		return shared.NewDecodeParamError(err.Error())
+	}
+	if len(obj) > 1 {
+		return fmt.Errorf("metricsArgs needs 0, 1 arguments")
+	}
+	// default values when not provided
+	if len(obj) >= 1 && obj[0] != nil {
+		if value, ok := obj[0].(bool); !ok {
+			return fmt.Errorf("invalid argument %v", reflect.TypeOf(obj[0]))
+		} else {
+			args.Raw = value
+		}
+	}
+	return nil
+}

rpc/api/debug_js.go

@@ -46,15 +46,17 @@ web3._extend({
 			params: 1,
 			inputFormatter: [web3._extend.formatters.formatInputInt],
 			outputFormatter: function(obj) { return obj; }
+		}),
+		new web3._extend.Method({
+			name: 'metrics',
+			call: 'debug_metrics',
+			params: 1,
+			inputFormatter: [web3._extend.formatters.formatInputBool],
+			outputFormatter: function(obj) { return obj; }
 		})
 	],
 	properties:
 	[
-		new web3._extend.Property({
-			name: 'metrics',
-			getter: 'debug_metrics',
-			outputFormatter: function(obj) { return obj; }
-		})
 	]
 });
 `

rpc/xeth.go

+package rpc
+
+import (
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"sync/atomic"
+
+	"github.com/ethereum/go-ethereum/rpc/comms"
+	"github.com/ethereum/go-ethereum/rpc/shared"
+)
+
+// Xeth is a native API interface to a remote node.
+type Xeth struct {
+	client comms.EthereumClient
+	reqId  uint32
+}
+
+// NewXeth constructs a new native API interface to a remote node.
+func NewXeth(client comms.EthereumClient) *Xeth {
+	return &Xeth{
+		client: client,
+	}
+}
+
+// Call invokes a method with the given parameters are the remote node.
+func (self *Xeth) Call(method string, params []interface{}) (map[string]interface{}, error) {
+	// Assemble the json RPC request
+	data, err := json.Marshal(params)
+	if err != nil {
+		return nil, err
+	}
+	req := &shared.Request{
+		Id:      atomic.AddUint32(&self.reqId, 1),
+		Jsonrpc: "2.0",
+		Method:  method,
+		Params:  data,
+	}
+	// Send the request over and process the response
+	if err := self.client.Send(req); err != nil {
+		return nil, err
+	}
+	res, err := self.client.Recv()
+	if err != nil {
+		return nil, err
+	}
+	value, ok := res.(map[string]interface{})
+	if !ok {
+		return nil, fmt.Errorf("Invalid response type: have %v, want %v", reflect.TypeOf(res), reflect.TypeOf(make(map[string]interface{})))
+	}
+	return value, nil
+}