From e5b820c47b9343d3801e1ebbeb4a8f40843ea87c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 24 Jun 2015 14:38:58 +0300 Subject: cmd/geth, rpc/api: extend metrics API, add a basic monitor command --- cmd/geth/main.go | 10 +++ cmd/geth/monitorcmd.go | 180 +++++++++++++++++++++++++++++++++++++++++++++++++ rpc/api/debug.go | 92 +++++++++++++++++-------- rpc/api/debug_args.go | 24 +++++++ rpc/api/debug_js.go | 12 ++-- rpc/xeth.go | 52 ++++++++++++++ 6 files changed, 338 insertions(+), 32 deletions(-) create mode 100644 cmd/geth/monitorcmd.go create mode 100644 rpc/xeth.go diff --git a/cmd/geth/main.go b/cmd/geth/main.go index 963aced15..f7b4810cd 100644 --- a/cmd/geth/main.go +++ b/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. `, }, { diff --git a/cmd/geth/monitorcmd.go b/cmd/geth/monitorcmd.go new file mode 100644 index 000000000..2989800f9 --- /dev/null +++ b/cmd/geth/monitorcmd.go @@ -0,0 +1,180 @@ +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 +} diff --git a/rpc/api/debug.go b/rpc/api/debug.go index 871786c6f..45c99f295 100644 --- a/rpc/api/debug.go +++ b/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,35 +206,69 @@ func (self *debugApi) Metrics(req *shared.Request) (interface{}, error) { } name = parts[len(parts)-1] - // Fill the counter with the metric details - switch metric := metric.(type) { - case metrics.Meter: - root[name] = map[string]interface{}{ - "Avg01Min": format(metric.Rate1()*60, metric.Rate1()), - "Avg05Min": format(metric.Rate5()*300, metric.Rate5()), - "Avg15Min": format(metric.Rate15()*900, metric.Rate15()), - "Total": format(float64(metric.Count()), metric.RateMean()), + // 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" } - - case metrics.Timer: - root[name] = map[string]interface{}{ - "Avg01Min": format(metric.Rate1()*60, metric.Rate1()), - "Avg05Min": format(metric.Rate5()*300, metric.Rate5()), - "Avg15Min": format(metric.Rate15()*900, metric.Rate15()), - "Count": format(float64(metric.Count()), metric.RateMean()), - "Maximum": time.Duration(metric.Max()).String(), - "Minimum": time.Duration(metric.Min()).String(), - "Percentile": map[string]interface{}{ - "20": time.Duration(metric.Percentile(0.2)).String(), - "50": time.Duration(metric.Percentile(0.5)).String(), - "80": time.Duration(metric.Percentile(0.8)).String(), - "95": time.Duration(metric.Percentile(0.95)).String(), - "99": time.Duration(metric.Percentile(0.99)).String(), - }, + } else { + switch metric := metric.(type) { + case metrics.Meter: + root[name] = map[string]interface{}{ + "Avg01Min": format(metric.Rate1()*60, metric.Rate1()), + "Avg05Min": format(metric.Rate5()*300, metric.Rate5()), + "Avg15Min": format(metric.Rate15()*900, metric.Rate15()), + "Total": format(float64(metric.Count()), metric.RateMean()), + } + + case metrics.Timer: + root[name] = map[string]interface{}{ + "Avg01Min": format(metric.Rate1()*60, metric.Rate1()), + "Avg05Min": format(metric.Rate5()*300, metric.Rate5()), + "Avg15Min": format(metric.Rate15()*900, metric.Rate15()), + "Count": format(float64(metric.Count()), metric.RateMean()), + "Maximum": time.Duration(metric.Max()).String(), + "Minimum": time.Duration(metric.Min()).String(), + "Percentile": map[string]interface{}{ + "20": time.Duration(metric.Percentile(0.2)).String(), + "50": time.Duration(metric.Percentile(0.5)).String(), + "80": time.Duration(metric.Percentile(0.8)).String(), + "95": time.Duration(metric.Percentile(0.95)).String(), + "99": time.Duration(metric.Percentile(0.99)).String(), + }, + } + + default: + root[name] = "Unknown metric type" } - - default: - root[name] = "Unknown metric type" } }) return counters, nil diff --git a/rpc/api/debug_args.go b/rpc/api/debug_args.go index b9b5aa27e..b72fb03ae 100644 --- a/rpc/api/debug_args.go +++ b/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 +} diff --git a/rpc/api/debug_js.go b/rpc/api/debug_js.go index e48e4df06..bd3a6dfb2 100644 --- a/rpc/api/debug_js.go +++ b/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; } - }) ] }); ` diff --git a/rpc/xeth.go b/rpc/xeth.go new file mode 100644 index 000000000..b3e844380 --- /dev/null +++ b/rpc/xeth.go @@ -0,0 +1,52 @@ +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 +} -- cgit v1.2.3