cmd, dashboard: some starter mistakes fixed

This commit is contained in:
Kurkó Mihály 2017-07-07 15:16:37 +03:00
parent 4c31d4b296
commit f1ca5702f9
6 changed files with 125 additions and 104 deletions

View file

@ -109,17 +109,13 @@ func defaultNodeConfig() node.Config {
return cfg return cfg
} }
func defaultDashboardConfig() dashboard.Config {
return dashboard.DefaultConfig
}
func makeConfigNode(ctx *cli.Context) (*node.Node, gethConfig) { func makeConfigNode(ctx *cli.Context) (*node.Node, gethConfig) {
// Load defaults. // Load defaults.
cfg := gethConfig{ cfg := gethConfig{
Eth: eth.DefaultConfig, Eth: eth.DefaultConfig,
Shh: whisper.DefaultConfig, Shh: whisper.DefaultConfig,
Node: defaultNodeConfig(), Node: defaultNodeConfig(),
Dashboard: defaultDashboardConfig(), Dashboard: dashboard.DefaultConfig,
} }
// Load config file. // Load config file.
@ -164,7 +160,6 @@ func makeFullNode(ctx *cli.Context) *node.Node {
if ctx.GlobalBool(utils.DashboardEnabledFlag.Name) { if ctx.GlobalBool(utils.DashboardEnabledFlag.Name) {
utils.RegisterDashboardService(stack, &cfg.Dashboard) utils.RegisterDashboardService(stack, &cfg.Dashboard)
} }
// Whisper must be explicitly enabled by specifying at least 1 whisper flag or in dev mode // Whisper must be explicitly enabled by specifying at least 1 whisper flag or in dev mode
shhEnabled := enableWhisper(ctx) shhEnabled := enableWhisper(ctx)
shhAutoEnabled := !ctx.GlobalIsSet(utils.WhisperEnabledFlag.Name) && ctx.GlobalIsSet(utils.DevModeFlag.Name) shhAutoEnabled := !ctx.GlobalIsSet(utils.WhisperEnabledFlag.Name) && ctx.GlobalIsSet(utils.DevModeFlag.Name)

View file

@ -1056,7 +1056,7 @@ func RegisterEthService(stack *node.Node, cfg *eth.Config) {
// RegisterDashboardService adds a dashboard to the stack. // RegisterDashboardService adds a dashboard to the stack.
func RegisterDashboardService(stack *node.Node, cfg *dashboard.Config) { func RegisterDashboardService(stack *node.Node, cfg *dashboard.Config) {
stack.Register(func(ctx *node.ServiceContext) (node.Service, error) { stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
return dashboard.NewDashboard(cfg) return dashboard.New(cfg)
}) })
} }

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View file

@ -23,10 +23,9 @@ var DefaultConfig = Config{
Host: "localhost", Host: "localhost",
Port: 8080, Port: 8080,
Refresh: time.Second, Refresh: time.Second,
Assets: "",
} }
//Config is the config of the dashboard // Config is the config of the dashboard
type Config struct { type Config struct {
// Host is the host interface on which to start the dashboard server. If this // Host is the host interface on which to start the dashboard server. If this
// field is empty, no dashboard will be started. // field is empty, no dashboard will be started.
@ -37,6 +36,10 @@ type Config struct {
// for ephemeral nodes). // for ephemeral nodes).
Port int `toml:",omitempty"` Port int `toml:",omitempty"`
Refresh time.Duration // Refresh is the refresh rate of the data updates, the data will be collected this often
Assets string Refresh time.Duration `toml:",omitempty"`
// Assets offers a possibility to manually set the dashboard website's location on the server side
// useful at debugging - avoids the repeated generation of the binary
Assets string `toml:",omitempty"`
} }

View file

@ -35,8 +35,8 @@ import (
) )
const ( const (
procSampleLimit = 200 processorSampleLimit = 200
memSampleLimit = 200 memorySampleLimit = 200
) )
var ( var (
@ -50,10 +50,12 @@ type dashboard struct {
index []byte // Index page to serve up on the web index []byte // Index page to serve up on the web
conns []*client // Currently live websocket connections conns []*client // Currently live websocket connections
mtrcs *mtrcs `json:",omitempty"` Metrics *metricSamples `json:"metrics,omitempty"`
stats *status `json:",omitempty"` Stats *status `json:"stats,omitempty"`
lock sync.RWMutex // Lock protecting the dashboard's internals lock sync.RWMutex // Lock protecting the dashboard's internals
quit chan struct{} // Channel used for graceful exit
} }
type client struct { type client struct {
@ -61,27 +63,29 @@ type client struct {
logger log.Logger // Logger for the particular live websocket connection logger log.Logger // Logger for the particular live websocket connection
} }
type mtrcs struct { type metricSamples struct {
Processor []*data `json:"proc,omitempty"` Processor []*data `json:"processor,omitempty"`
Memory []*data `json:"mem,omitempty"` Memory []*data `json:"memory,omitempty"`
} }
type data struct { type data struct {
T int `json:"time,omitempty"` Time time.Time `json:"time,omitempty"`
Value float64 `json:"value,omitempty"` Value float64 `json:"value,omitempty"`
} }
type status struct { type status struct {
Peers int Peers int `json:"peers,omitempty"`
Block int Block int `json:"block,omitempty"`
} }
func NewDashboard(config *Config) (*dashboard, error) { // New creates a new dashboard instance with the given configuration
log.Trace("NewDashboard() called") func New(config *Config) (*dashboard, error) {
//log.Trace("NewDashboard() called")
dashboard := &dashboard{ dashboard := &dashboard{
config: config, config: config,
mtrcs: &mtrcs{}, Metrics: &metricSamples{},
quit: make(chan struct{}),
} }
if config.Assets == "" { if config.Assets == "" {
@ -91,9 +95,10 @@ func NewDashboard(config *Config) (*dashboard, error) {
} }
website := new(bytes.Buffer) website := new(bytes.Buffer)
// set the sample limits for the client
if err = template.Must(template.New("").Parse(string(tmpl))).Execute(website, map[string]interface{}{ if err = template.Must(template.New("").Parse(string(tmpl))).Execute(website, map[string]interface{}{
"procSampleLimit": procSampleLimit, "processorSampleLimit": processorSampleLimit,
"memSampleLimit": memSampleLimit, "memorySampleLimit": memorySampleLimit,
}); err != nil { }); err != nil {
log.Crit("Failed to render the dashboard template", "err", err) log.Crit("Failed to render the dashboard template", "err", err)
} }
@ -102,17 +107,20 @@ func NewDashboard(config *Config) (*dashboard, error) {
return dashboard, nil return dashboard, nil
} }
//TODO case: DashboardAssetsFlag is set //TODO (kurkomisi): case DashboardAssetsFlag is set
//dashboard.index = ioutil.ReadFile() //dashboard.index = ioutil.ReadFile()
return dashboard, nil return dashboard, nil
} }
// Protocols is a meaningless implementation of node.Service
func (db *dashboard) Protocols() []p2p.Protocol { return nil } func (db *dashboard) Protocols() []p2p.Protocol { return nil }
// APIs is a meaningless implementation of node.Service
func (db *dashboard) APIs() []rpc.API { return nil } func (db *dashboard) APIs() []rpc.API { return nil }
// Start implements node.Service, starting the data collection thread and the listening server of the dashboard
func (db *dashboard) Start(server *p2p.Server) error { func (db *dashboard) Start(server *p2p.Server) error {
log.Trace("Start() called", "config", db.config) //log.Trace("Start() called", "config", db.config)
go db.collectData() go db.collectData()
@ -126,7 +134,7 @@ func (db *dashboard) Start(server *p2p.Server) error {
db.listener = listener db.listener = listener
go func() { go func() {
log.Trace("Starting server...") //log.Trace("Starting server...")
if err := http.Serve(listener, nil); err != nil { if err := http.Serve(listener, nil); err != nil {
log.Warn("Server failed", "err", err) log.Warn("Server failed", "err", err)
@ -136,16 +144,21 @@ func (db *dashboard) Start(server *p2p.Server) error {
return nil return nil
} }
// Stop implements node.Service, stopping the data collection thread and the connection listener of the dashboard
func (db *dashboard) Stop() error { func (db *dashboard) Stop() error {
log.Trace("Terminating dashboard...") //log.Trace("Terminating dashboard...")
var err error var err error
// Close the connection listener
db.lock.Lock() db.lock.Lock()
if err = db.listener.Close(); err != nil { if err = db.listener.Close(); err != nil {
log.Warn("Failed to close listener", "err", err) log.Warn("Failed to close listener", "err", err)
} }
// Notifies collectData and apiHandler
close(db.quit)
for _, c := range db.conns { for _, c := range db.conns {
if err := c.conn.Close(); err != nil { if err := c.conn.Close(); err != nil {
c.logger.Warn("Failed to close connection", "err", err) c.logger.Warn("Failed to close connection", "err", err)
@ -159,15 +172,15 @@ func (db *dashboard) Stop() error {
// webHandler handles all non-api requests, simply flattening and returning the dashboard website. // webHandler handles all non-api requests, simply flattening and returning the dashboard website.
func (db *dashboard) webHandler(w http.ResponseWriter, r *http.Request) { func (db *dashboard) webHandler(w http.ResponseWriter, r *http.Request) {
log.Trace("webHandler() called") //log.Trace("webHandler() called")
//TODO not only index //TODO (kurkomisi): not only index
w.Write(db.index) w.Write(db.index)
} }
// apiHandler handles requests for dashboard // apiHandler handles requests for dashboard
func (db *dashboard) apiHandler(conn *websocket.Conn) { func (db *dashboard) apiHandler(conn *websocket.Conn) {
log.Trace("apiHandler() called") //log.Trace("apiHandler() called")
client := &client{ client := &client{
conn: conn, conn: conn,
@ -179,89 +192,97 @@ func (db *dashboard) apiHandler(conn *websocket.Conn) {
db.conns = append(db.conns, client) db.conns = append(db.conns, client)
db.lock.Unlock() db.lock.Unlock()
defer func() { db.sendHistory(client)
client.logger.Trace("Connection interrupted")
closed := make(chan bool)
go func() {
select {
case <-db.quit:
//client.logger.Trace("apiHandler closed")
case <-closed:
//client.logger.Trace("Connection interrupted")
db.lock.Lock() db.lock.Lock()
for i, c := range db.conns { for i, c := range db.conns {
if c.conn == client.conn { if c.conn == client.conn {
if err := c.conn.Close(); err != nil {
c.logger.Warn("Failed to close connection", "err", err)
}
db.conns = append(db.conns[:i], db.conns[i+1:]...) db.conns = append(db.conns[:i], db.conns[i+1:]...)
break break
} }
} }
db.lock.Unlock() db.lock.Unlock()
}
}() }()
db.sendHistory(client)
for { for {
var msg struct { fail := []byte{}
text string `json:"text"` if _, err := conn.Read(fail); err != nil {
} closed <- true
if err := websocket.JSON.Receive(conn, &msg); err != nil {
client.logger.Warn("Receive failed", "err", err)
return return
} }
// Ignore any message
} }
} }
// collectData collects the required data to plot on the dashboard // collectData collects the required data to plot on the dashboard
func (db *dashboard) collectData() { func (db *dashboard) collectData() {
log.Trace("collectData() called") //log.Trace("collectData() called")
for { for {
now := time.Now().Second() select {
case <-db.quit:
//log.Trace("collectData closed")
return
case <-time.After(db.config.Refresh):
now := time.Now()
traffic := metrics.DefaultRegistry.Get("p2p/InboundTraffic").(metrics.Meter).Rate1() traffic := metrics.DefaultRegistry.Get("p2p/InboundTraffic").(metrics.Meter).Rate1()
traffic = traffic * traffic traffic = traffic * traffic
//if traffic != 0 { //if traffic != 0 {
// traffic = math.Log(traffic) // traffic = math.Log(traffic)
//} //}
memInuse := metrics.DefaultRegistry.Get("system/memory/inuse").(metrics.Meter).Rate1() memoryInuse := metrics.DefaultRegistry.Get("system/memory/inuse").(metrics.Meter).Rate1()
//if memInuse != 0 { //if memoryInuse != 0 {
// memInuse = math.Log(memInuse) // memoryInuse = math.Log(memoryInuse)
//} //}
traff := &data{ traff := &data{
T: now, Time: now,
Value: traffic, Value: traffic,
} }
mem := &data{ memory := &data{
T: now, Time: now,
Value: memInuse, Value: memoryInuse,
} }
db.update(traff, mem) //TODO (kurkomisi): do I need to ensure the correct order?
go db.update(traff, memory)
time.Sleep(db.config.Refresh) }
} }
} }
// update updates the dashboards through the live websocket connections // update updates the dashboards through the live websocket connections
func (db *dashboard) update(proc *data, mem *data) { func (db *dashboard) update(processor *data, memory *data) {
//log.Trace("update() called") //log.Trace("update() called")
db.lock.Lock()
defer db.lock.Unlock()
// if the samples' # exceeds the limit, just remove the first element // if the samples' # exceeds the limit, just remove the first element
first := 0 first := 0
if len(db.mtrcs.Processor) == procSampleLimit { if len(db.Metrics.Processor) == processorSampleLimit {
first = 1 first = 1
} }
db.mtrcs.Processor = append(db.mtrcs.Processor[first:], proc) db.Metrics.Processor = append(db.Metrics.Processor[first:], processor)
first = 0 first = 0
if len(db.mtrcs.Memory) == memSampleLimit { if len(db.Metrics.Memory) == memorySampleLimit {
first = 1 first = 1
} }
db.mtrcs.Memory = append(db.mtrcs.Memory[first:], mem) db.Metrics.Memory = append(db.Metrics.Memory[first:], memory)
for _, c := range db.conns { for _, c := range db.conns {
//c.logger.Trace("Updating dashboard...") //c.logger.Trace("Updating dashboard...")
msg := &map[string]interface{}{ msg := &map[string]interface{}{
"proc": proc, "processor": processor,
"mem": mem, "memory": memory,
} }
if err := websocket.JSON.Send(c.conn, msg); err != nil { if err := websocket.JSON.Send(c.conn, msg); err != nil {
c.logger.Warn("Failed to update dashboard", "msg", msg, "err", err) c.logger.Warn("Failed to update dashboard", "msg", msg, "err", err)
@ -270,10 +291,12 @@ func (db *dashboard) update(proc *data, mem *data) {
} }
// sendHistory sends the past data through a newly registered websocket connection
func (db *dashboard) sendHistory(c *client) { func (db *dashboard) sendHistory(c *client) {
c.logger.Trace("Sending history...") //c.logger.Trace("Sending history...")
msg := &map[string]interface{}{ msg := &map[string]interface{}{
"mtrcs": db.mtrcs, "metrics": db.Metrics,
} }
if err := websocket.JSON.Send(c.conn, msg); err != nil { if err := websocket.JSON.Send(c.conn, msg); err != nil {
c.logger.Warn("Failed to send history", "err", err) c.logger.Warn("Failed to send history", "err", err)