cmd, dashboard, internal, log, node: create log files, visualize log records

cmd/geth/main.go

@@ -185,7 +185,12 @@ func init() {
 
 	app.Before = func(ctx *cli.Context) error {
 		runtime.GOMAXPROCS(runtime.NumCPU())
-		if err := debug.Setup(ctx, ctx.GlobalBool(utils.DashboardEnabledFlag.Name), utils.DataDirFlag.Value.Value); err != nil {
+
+		disklogs := ""
+		if ctx.GlobalBool(utils.DashboardEnabledFlag.Name) {
+			disklogs = (&node.Config{DataDir: utils.MakeDataDir(ctx)}).ResolvePath("logs")
+		}
+		if err := debug.Setup(ctx, disklogs); err != nil {
 			return err
 		}
 		// Start system runtime metrics collection

cmd/swarm/main.go

@@ -46,6 +46,7 @@ import (
 	swarmmetrics "github.com/ethereum/go-ethereum/swarm/metrics"
 
 	"gopkg.in/urfave/cli.v1"
+	"path/filepath"
 )
 
 const clientIdentifier = "swarm"
@@ -363,7 +364,11 @@ DEPRECATED: use 'swarm db clean'.
 	app.Flags = append(app.Flags, swarmmetrics.Flags...)
 	app.Before = func(ctx *cli.Context) error {
 		runtime.GOMAXPROCS(runtime.NumCPU())
-		if err := debug.Setup(ctx, ctx.GlobalBool(utils.DashboardEnabledFlag.Name), utils.DataDirFlag.Value.Value); err != nil {
+		disklogs := ""
+		if ctx.GlobalBool(utils.DashboardEnabledFlag.Name) {
+			disklogs = (&node.Config{DataDir: utils.MakeDataDir(ctx)}).ResolvePath("logs")
+		}
+		if err := debug.Setup(ctx, disklogs); err != nil {
 			return err
 		}
 		swarmmetrics.Setup(ctx)

cmd/utils/flags.go

@@ -1145,7 +1145,7 @@ 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)
+		return dashboard.New(cfg, commit, ctx.ResolvePath("logs"))
 	})
 }
 

dashboard/assets/components/Body.jsx

@@ -37,6 +37,8 @@ export type Props = {
 	active: string,
 	content: Content,
 	shouldUpdate: Object,
+	send: (string) => void,
+	logs: () => Object,
 };
 
 // Body renders the body of the dashboard.
@@ -52,6 +54,8 @@ class Body extends Component<Props> {
 					active={this.props.active}
 					content={this.props.content}
 					shouldUpdate={this.props.shouldUpdate}
+					send={this.props.send}
+					logs={this.props.logs}
 				/>
 			</div>
 		);

dashboard/assets/components/Dashboard.jsx

@@ -18,12 +18,13 @@
 
 import React, {Component} from 'react';
 
+import List, {ListItem} from 'material-ui/List';
 import withStyles from 'material-ui/styles/withStyles';
 
 import Header from './Header';
 import Body from './Body';
 import {MENU} from '../common';
-import type {Content} from '../types/content';
+import type {Content, Record, Chunk} from '../types/content';
 
 // deepUpdate updates an object corresponding to the given update data, which has
 // the shape of the same structure as the original object. updater also has the same
@@ -75,6 +76,104 @@ const appender = <T>(limit: number, mapper = replacer) => (update: Array<T>, pre
 	...update.map(sample => mapper(sample)),
 ].slice(-limit);
 
+// fieldPadding is a global map with maximum field value lengths seen until now
+// to allow padding log contexts in a bit smarter way.
+const fieldPadding = new Map();
+
+// createLogChunk creates an HTML formatted object, which displays the given array similarly to
+// the server side terminal.
+const createLogChunk = (arr: Array<Record>) => {
+	let content = '';
+	arr.forEach((record) => {
+		let {t, lvl, msg, ctx} = record;
+		let color = '#ce3c23';
+		switch (lvl) {
+		case 'trace':
+		case 'trce':
+			lvl = 'TRACE';
+			color = '#3465a4';
+			break;
+		case 'debug':
+		case 'dbug':
+			lvl = 'DEBUG';
+			color = '#3d989b';
+			break;
+		case 'info':
+			lvl = 'INFO&nbsp;';
+			color = '#4c8f0f';
+			break;
+		case 'warn':
+			lvl = 'WARN&nbsp;';
+			color = '#b79a22';
+			break;
+		case 'error':
+		case 'eror':
+			lvl = 'ERROR';
+			color = '#754b70';
+			break;
+		case 'crit':
+			lvl = 'CRIT&nbsp;';
+			color = '#ce3c23';
+			break;
+		default:
+			lvl = '';
+		}
+		if (lvl === '' || typeof t !== 'string' || t.length < 19 || typeof msg !== 'string' || !Array.isArray(ctx)) {
+			content += `<span style="color:${color}">Invalid log record</span><br />`;
+			return;
+		}
+		if (ctx.length > 0) {
+			msg += '&nbsp;'.repeat(Math.max(40 - msg.length, 0));
+		}
+		// Time format: 2006-01-02T15:04:05-0700 -> 01-02|15:04:05
+		content += `<span style="color:${color}">${lvl}</span>[${t.substr(5, 5)}|${t.substr(11, 8)}] ${msg}`;
+
+		for (let i = 0; i < ctx.length; i += 2) {
+			const key = ctx[i];
+			const value = ctx[i + 1];
+			let padding = fieldPadding.get(key);
+			if (typeof padding === 'undefined' || padding < value.length) {
+				padding = value.length;
+				fieldPadding.set(key, padding);
+			}
+			content += ` <span style="color:${color}">${key}</span>=${value}${'&nbsp;'.repeat(padding - value.length)}`;
+		}
+		content += '<br />';
+	});
+	return content;
+};
+
+// logAppender is a state updater function, which appends the new log chunks to the existing ones.
+// In case the prev chunk array's last element doesn't have limit number of log record elements,
+// it will be extended.
+const logAppender = (limit: number) => (update: Array<Record>, prev: Array<Chunk>) => {
+	const newChunks = [];
+	let first = 0;
+	let last = 0;
+	let extended = 0;
+	if (prev.length > 0 && prev[prev.length - 1].len < limit) {
+		extended = 1;
+		const l = Math.min(limit - prev[prev.length - 1].len, update.length);
+		newChunks.push({
+			content: prev[prev.length - 1].content + createLogChunk(update.slice(0, l)),
+			t:       prev[prev.length - 1].t,
+			len:     prev[prev.length - 1].len + l,
+		});
+		first = l;
+		last = l;
+	}
+	while (last < update.length) {
+		last = Math.min(update.length, last + limit);
+		newChunks.push({
+			content: createLogChunk(update.slice(first, last)),
+			t:       update[first].t,
+			len:     last - first,
+		});
+		first += limit;
+	}
+	return [...prev.slice(0, prev.length - extended), ...newChunks];
+};
+
 // defaultContent is the initial value of the state content.
 const defaultContent: Content = {
 	general: {
@@ -96,7 +195,7 @@ const defaultContent: Content = {
 		diskWrite:      [],
 	},
 	logs:    {
-		log: [],
+		chunk: [],
 	},
 };
 
@@ -123,7 +222,7 @@ const updaters = {
 		diskWrite:      appender(200),
 	},
 	logs: {
-		log: appender(200),
+		chunk: logAppender(50),
 	},
 };
 
@@ -137,6 +236,10 @@ const styles = {
 		zIndex:   1,
 		overflow: 'hidden',
 	},
+	logChunk: {
+		color:      'white',
+		fontFamily: 'monospace',
+	},
 };
 
 // themeStyles returns the styles generated from the theme for the component.
@@ -155,6 +258,7 @@ type State = {
 	sideBar: boolean, // true if the sidebar is opened
 	content: Content, // the visualized data
 	shouldUpdate: Object, // labels for the components, which need to re-render based on the incoming message
+	server: ?WebSocket,
 };
 
 // Dashboard is the main component, which renders the whole page, makes connection with the server and
@@ -167,6 +271,7 @@ class Dashboard extends Component<Props, State> {
 			sideBar:      true,
 			content:      defaultContent,
 			shouldUpdate: {},
+			server:       null,
 		};
 	}
 
@@ -181,7 +286,7 @@ class Dashboard extends Component<Props, State> {
 		// PROD is defined by webpack.
 		const server = new WebSocket(`${((window.location.protocol === 'https:') ? 'wss://' : 'ws://')}${PROD ? window.location.host : 'localhost:8080'}/api`);
 		server.onopen = () => {
-			this.setState({content: defaultContent, shouldUpdate: {}});
+			this.setState({content: defaultContent, shouldUpdate: {}, server});
 		};
 		server.onmessage = (event) => {
 			const msg: $Shape<Content> = JSON.parse(event.data);
@@ -192,10 +297,18 @@ class Dashboard extends Component<Props, State> {
 			this.update(msg);
 		};
 		server.onclose = () => {
+			this.setState({server: null});
 			setTimeout(this.reconnect, 3000);
 		};
 	};
 
+	// server can be accessed only through this function for safety reasons.
+	send = (msg: string) => {
+		if (this.state.server != null) {
+			this.state.server.send(msg);
+		}
+	};
+
 	// update updates the content corresponding to the incoming message.
 	update = (msg: $Shape<Content>) => {
 		this.setState(prevState => ({
@@ -214,6 +327,18 @@ class Dashboard extends Component<Props, State> {
 		this.setState(prevState => ({sideBar: !prevState.sideBar}));
 	};
 
+	// logsHTML visualizes the log chunks. It is more efficient to insert pure HTML into the component, than
+	// to create individual component for each log record and track them all. It also postpones the OOM issue.
+	logsHTML = () => (
+		<List>
+			{this.state.content.logs.chunk.map((c, index) => (
+				<ListItem key={index}>
+					<div style={styles.logChunk} dangerouslySetInnerHTML={{__html: c.content}} />
+				</ListItem>
+			))}
+		</List>
+	);
+
 	render() {
 		return (
 			<div className={this.props.classes.dashboard} style={styles.dashboard}>
@@ -226,6 +351,8 @@ class Dashboard extends Component<Props, State> {
 					active={this.state.active}
 					content={this.state.content}
 					shouldUpdate={this.state.shouldUpdate}
+					send={this.send}
+					logs={this.logsHTML}
 				/>
 			</div>
 		);

dashboard/assets/components/Logs.jsx

@@ -0,0 +1,36 @@
+// @flow
+
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+import React, {Component} from 'react';
+
+export type Props = {
+    logs: () => Object,
+};
+
+// Logs renders the log page.
+class Logs extends Component<Props> {
+	render() {
+		return (
+			<div >
+				{this.props.logs()}
+			</div>
+		);
+	}
+}
+
+export default Logs;

dashboard/assets/components/Main.jsx

@@ -21,6 +21,7 @@ import React, {Component} from 'react';
 import withStyles from 'material-ui/styles/withStyles';
 
 import {MENU} from '../common';
+import Logs from './Logs';
 import Footer from './Footer';
 import type {Content} from '../types/content';
 
@@ -50,10 +51,32 @@ export type Props = {
 	active: string,
 	content: Content,
 	shouldUpdate: Object,
+    send: (string) => void,
+	logs: () => Object,
 };
 
 // Main renders the chosen content.
 class Main extends Component<Props> {
+	handleScroll = () => {
+		if (typeof this.container !== 'undefined') {
+			// console.log(this.container.scrollTop, this.container.scrollHeight);
+			if (this.container.scrollTop === 0) {
+				// this.props.send(JSON.stringify({Logs: {Time: '2018-04-11T12:48:18.181274193+03:00'}}));
+				console.log("Top");
+			}
+			if (this.container.scrollHeight - this.container.scrollTop === this.container.clientHeight) {
+				console.log("Bottom");
+				// this.container.scrollTop = 0;
+			}
+		}
+	};
+
+	componentDidUpdate() {
+		// if (typeof this.container !== 'undefined') {
+		// 	this.container.scrollTop = this.container.scrollHeight - this.container.clientHeight;
+		// }
+	}
+
 	render() {
 		const {
 			classes, active, content, shouldUpdate,
@@ -69,12 +92,19 @@ class Main extends Component<Props> {
 			children = <div>Work in progress.</div>;
 			break;
 		case MENU.get('logs').id:
-			children = <div>{content.logs.log.map((log, index) => <div key={index}>{log}</div>)}</div>;
+			children = <Logs logs={this.props.logs} />;
 		}
 
 		return (
 			<div style={styles.wrapper}>
-				<div className={classes.content} style={styles.content}>{children}</div>
+				<div
+					className={classes.content}
+					style={styles.content}
+					ref={(ref) => { this.container = ref; }}
+					onScroll={this.handleScroll}
+				>
+					{children}
+				</div>
 				<Footer
 					general={content.general}
 					system={content.system}

dashboard/assets/types/content.jsx

@@ -65,6 +65,19 @@ export type System = {
     diskWrite:      ChartEntries,
 };
 
-export type Logs = {
-	log: Array<string>,
+export type Record = {
+    t:   Object,
+    lvl: Object,
+    msg: string,
+    ctx: Array<string>
+};
+
+export type Chunk = {
+    content: string,
+    t:       string,
+    len:     int,
+};
+
+export type Logs = {
+    chunk: Array<Chunk>,
 };

dashboard/dashboard.go

@@ -32,13 +32,21 @@ import (
 	"sync/atomic"
 	"time"
 
+	"encoding/json"
 	"github.com/elastic/gosigar"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/fsnotify/fsnotify"
 	"golang.org/x/net/websocket"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"strings"
 )
 
 const (
@@ -60,10 +68,11 @@ type Dashboard struct {
 
 	listener net.Listener
 	conns    map[uint32]*client // Currently live websocket connections
-	charts   *SystemMessage
-	commit   string
+	history  *Message
 	lock     sync.RWMutex // Lock protecting the dashboard's internals
 
+	logdir string
+
 	quit chan chan error // Channel used for graceful exit
 	wg   sync.WaitGroup
 }
@@ -71,18 +80,27 @@ type Dashboard struct {
 // client represents active websocket connection with a remote browser.
 type client struct {
 	conn   *websocket.Conn // Particular live websocket connection
-	msg    chan Message    // Message queue for the update messages
+	msg    chan *Message   // Message queue for the update messages
 	logger log.Logger      // Logger for the particular live websocket connection
 }
 
 // New creates a new dashboard instance with the given configuration.
-func New(config *Config, commit string) (*Dashboard, error) {
+func New(config *Config, commit string, logdir string) (*Dashboard, error) {
 	now := time.Now()
-	db := &Dashboard{
+	versionMeta := ""
+	if len(params.VersionMeta) > 0 {
+		versionMeta = fmt.Sprintf(" (%s)", params.VersionMeta)
+	}
+	return &Dashboard{
 		conns:  make(map[uint32]*client),
 		config: config,
 		quit:   make(chan chan error),
-		charts: &SystemMessage{
+		history: &Message{
+			General: &GeneralMessage{
+				Commit:  commit,
+				Version: fmt.Sprintf("v%d.%d.%d%s", params.VersionMajor, params.VersionMinor, params.VersionPatch, versionMeta),
+			},
+			System: &SystemMessage{
 				ActiveMemory:   emptyChartEntries(now, activeMemorySampleLimit, config.Refresh),
 				VirtualMemory:  emptyChartEntries(now, virtualMemorySampleLimit, config.Refresh),
 				NetworkIngress: emptyChartEntries(now, networkIngressSampleLimit, config.Refresh),
@@ -92,9 +110,10 @@ func New(config *Config, commit string) (*Dashboard, error) {
 				DiskRead:       emptyChartEntries(now, diskReadSampleLimit, config.Refresh),
 				DiskWrite:      emptyChartEntries(now, diskWriteSampleLimit, config.Refresh),
 			},
-		commit: commit,
-	}
-	return db, nil
+			Logs: &LogsMessage{Chunk: json.RawMessage("[]")},
+		},
+		logdir: logdir,
+	}, nil
 }
 
 // emptyChartEntries returns a ChartEntry array containing limit number of empty samples.
@@ -120,7 +139,7 @@ func (db *Dashboard) Start(server *p2p.Server) error {
 
 	db.wg.Add(2)
 	go db.collectData()
-	go db.collectLogs() // In case of removing this line change 2 back to 1 in wg.Add.
+	go db.streamLogs()
 
 	http.HandleFunc("/", db.webHandler)
 	http.Handle("/api", websocket.Handler(db.apiHandler))
@@ -194,7 +213,7 @@ func (db *Dashboard) apiHandler(conn *websocket.Conn) {
 	id := atomic.AddUint32(&nextID, 1)
 	client := &client{
 		conn:   conn,
-		msg:    make(chan Message, 128),
+		msg:    make(chan *Message, 128),
 		logger: log.New("id", id),
 	}
 	done := make(chan struct{})
@@ -218,29 +237,10 @@ func (db *Dashboard) apiHandler(conn *websocket.Conn) {
 		}
 	}()
 
-	versionMeta := ""
-	if len(params.VersionMeta) > 0 {
-		versionMeta = fmt.Sprintf(" (%s)", params.VersionMeta)
-	}
-	// Send the past data.
-	client.msg <- Message{
-		General: &GeneralMessage{
-			Version: fmt.Sprintf("v%d.%d.%d%s", params.VersionMajor, params.VersionMinor, params.VersionPatch, versionMeta),
-			Commit:  db.commit,
-		},
-		System: &SystemMessage{
-			ActiveMemory:   db.charts.ActiveMemory,
-			VirtualMemory:  db.charts.VirtualMemory,
-			NetworkIngress: db.charts.NetworkIngress,
-			NetworkEgress:  db.charts.NetworkEgress,
-			ProcessCPU:     db.charts.ProcessCPU,
-			SystemCPU:      db.charts.SystemCPU,
-			DiskRead:       db.charts.DiskRead,
-			DiskWrite:      db.charts.DiskWrite,
-		},
-	}
-	// Start tracking the connection and drop at connection loss.
 	db.lock.Lock()
+	// Send the past data.
+	client.msg <- db.history.DeepCopy()
+	// Start tracking the connection and drop at connection loss.
 	db.conns[id] = client
 	db.lock.Unlock()
 	defer func() {
@@ -249,18 +249,80 @@ func (db *Dashboard) apiHandler(conn *websocket.Conn) {
 		db.lock.Unlock()
 	}()
 	for {
-		fail := []byte{}
-		if _, err := conn.Read(fail); err != nil {
+		var r Request
+		err := websocket.JSON.Receive(conn, &r)
+		if err != nil {
 			close(done)
 			return
 		}
-		// Ignore all messages
+		if r.Logs != nil {
+			db.handleLogs(r.Logs, client) // TODO (kurkomisi): concurrent function call?
+		}
+	}
+}
+
+// handleLogs searches for the log file specified by the timestamp of the request, creates a JSON array out of it
+// and sends it to the requesting client.
+func (db *Dashboard) handleLogs(r *LogsRequest, c *client) {
+	files, err := ioutil.ReadDir(db.logdir)
+	if err != nil {
+		log.Warn("Failed to open logdir", "logdir", db.logdir, "err", err)
+		return
+	}
+	re := regexp.MustCompile(".log$")
+	valid := make([]string, len(files))
+	n := 0
+	for _, f := range files {
+		if f.Mode().IsRegular() && re.Match([]byte(f.Name())) {
+			valid[n] = f.Name()
+			n++
+		}
+	}
+	if len(valid) < 1 {
+		log.Warn("There isn't any log file in the logdir", "logdir", db.logdir)
+		return
+	}
+	timestamp := fmt.Sprintf("%s.log", strings.Replace(r.Time.Format("060102150405.00"), ".", "", 1))
+	i := sort.Search(len(valid), func(i int) bool {
+		return valid[i] >= timestamp
+	})
+	if i >= len(valid) {
+		i = len(valid) - 1
+	}
+	f, err := os.OpenFile(filepath.Join(db.logdir, valid[i]), os.O_RDONLY, 0644)
+	if err != nil {
+		log.Warn("Failed to open file", "name", valid[i], "err", err)
+		return
+	}
+	defer f.Close()
+	buf, err := ioutil.ReadAll(f)
+	last := -1
+	for i := 0; i < len(buf); i++ {
+		if buf[i] == '\n' {
+			buf[i] = ','
+			last = i
+		}
+	}
+	if last >= 0 {
+		b := make([]byte, last+2)
+		b[0] = '['
+		copy(b[1:], buf[:last])
+		b[last+1] = ']'
+
+		db.lock.Lock() // TODO (kurkomisi): Maybe create mutex for the client.
+		c.msg <- &Message{
+			Logs: &LogsMessage{
+				Chunk: b,
+			},
+		}
+		db.lock.Unlock()
 	}
 }
 
 // collectData collects the required data to plot on the dashboard.
 func (db *Dashboard) collectData() {
 	defer db.wg.Done()
+
 	systemCPUUsage := gosigar.Cpu{}
 	systemCPUUsage.Get()
 	var (
@@ -275,6 +337,8 @@ func (db *Dashboard) collectData() {
 
 		frequency = float64(db.config.Refresh / time.Second)
 		numCPU    = float64(runtime.NumCPU())
+
+		sys = db.history.System
 	)
 
 	for {
@@ -341,14 +405,14 @@ func (db *Dashboard) collectData() {
 				Time:  now,
 				Value: float64(deltaDiskWrite) / frequency,
 			}
-			db.charts.ActiveMemory = append(db.charts.ActiveMemory[1:], activeMemory)
-			db.charts.VirtualMemory = append(db.charts.VirtualMemory[1:], virtualMemory)
-			db.charts.NetworkIngress = append(db.charts.NetworkIngress[1:], networkIngress)
-			db.charts.NetworkEgress = append(db.charts.NetworkEgress[1:], networkEgress)
-			db.charts.ProcessCPU = append(db.charts.ProcessCPU[1:], processCPU)
-			db.charts.SystemCPU = append(db.charts.SystemCPU[1:], systemCPU)
-			db.charts.DiskRead = append(db.charts.DiskRead[1:], diskRead)
-			db.charts.DiskWrite = append(db.charts.DiskRead[1:], diskWrite)
+			sys.ActiveMemory = append(sys.ActiveMemory[1:], activeMemory)
+			sys.VirtualMemory = append(sys.VirtualMemory[1:], virtualMemory)
+			sys.NetworkIngress = append(sys.NetworkIngress[1:], networkIngress)
+			sys.NetworkEgress = append(sys.NetworkEgress[1:], networkEgress)
+			sys.ProcessCPU = append(sys.ProcessCPU[1:], processCPU)
+			sys.SystemCPU = append(sys.SystemCPU[1:], systemCPU)
+			sys.DiskRead = append(sys.DiskRead[1:], diskRead)
+			sys.DiskWrite = append(sys.DiskRead[1:], diskWrite)
 
 			db.sendToAll(&Message{
 				System: &SystemMessage{
@@ -366,24 +430,130 @@ func (db *Dashboard) collectData() {
 	}
 }
 
-// collectLogs collects and sends the logs to the active dashboards.
-func (db *Dashboard) collectLogs() {
+// streamLogs watches the file system, and when the logger writes the new log records into the files, picks them up,
+// then makes JSON array out of them and sends them to the clients.
+// This could be embedded into collectData, but they shouldn't depend on each other, and also cleaner this way.
+func (db *Dashboard) streamLogs() {
 	defer db.wg.Done()
 
-	id := 1
-	// TODO (kurkomisi): log collection comes here.
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		log.Warn("Failed to create fs watcher", "err", err)
+		return
+	}
+	defer watcher.Close()
+
+	files, err := ioutil.ReadDir(db.logdir)
+	if err != nil {
+		log.Warn("Failed to open logdir", "logdir", db.logdir, "err", err)
+		return
+	}
+	var (
+		opened *os.File // File descriptor for the opened active log file.
+		buf    []byte   // Contains the recently written log chunks, which are not sent to the clients yet.
+	)
+
+	// The log records are always written into the last file in alphabetical order, because of the timestamp.
+	re := regexp.MustCompile(".log$")
+	var i int
+	for i = len(files) - 1; i >= 0 && (!files[i].Mode().IsRegular() || !re.Match([]byte(files[i].Name()))); i-- {
+	}
+	if i >= 0 {
+		if opened, err = os.OpenFile(filepath.Join(db.logdir, files[i].Name()), os.O_RDONLY, 0644); err != nil {
+			log.Warn("Failed to open file", "name", files[i].Name(), "err", err)
+			return
+		}
+		if buf, err = ioutil.ReadAll(opened); err != nil {
+			log.Warn("Failed to read file", "name", opened.Name(), "err", err)
+			return
+		}
+	}
+
+	err = watcher.Add(db.logdir)
+	if err != nil {
+		log.Warn("Failed to add logdir to fs watcher", "logdir", db.logdir, "err", err)
+		return
+	}
+	change := fsnotify.Create | fsnotify.Remove | fsnotify.Rename
 	for {
 		select {
+		case event := <-watcher.Events:
+			switch {
+			// If new log file is opened.
+			case event.Op&change != 0:
+				if re.Match([]byte(event.Name)) && opened.Name() < event.Name {
+					if opened, err = os.OpenFile(event.Name, os.O_RDONLY, 0644); err != nil {
+						log.Warn("Failed to open file", "name", event.Name, "err", err)
+						return
+					}
+					db.lock.Lock()
+					db.history.Logs.Chunk = json.RawMessage("[]")
+					db.lock.Unlock()
+				}
+			// If new log records were written into the opened log file.
+			case event.Op&fsnotify.Write != 0:
+				if opened != nil {
+					chunk, err := ioutil.ReadAll(opened)
+					if err != nil {
+						log.Warn("Failed to read file", "name", opened.Name(), "err", err)
+						return
+					}
+					b := make([]byte, len(buf)+len(chunk))
+					copy(b, buf)
+					copy(b[len(buf):], chunk)
+					buf = b
+				}
+			}
+		case err := <-watcher.Errors:
+			if err != nil {
+				log.Warn("Fs watcher error", "err", err)
+			}
+			return
+		// Send log updates to the client.
+		case <-time.After(db.config.Refresh):
+			last := -1
+			for i := 0; i < len(buf); i++ {
+				if buf[i] == '\n' {
+					buf[i] = ','
+					last = i
+				}
+			}
+			if last >= 0 {
+				b := make([]byte, last+2)
+				b[0] = '['
+				copy(b[1:], buf[:last])
+				b[last+1] = ']'
+
+				db.sendToAll(&Message{
+					Logs: &LogsMessage{
+						Chunk: b,
+					},
+				})
+
+				b = make([]byte, len(db.history.Logs.Chunk)+last+1)
+				// Cut the ']' from the end in order to concatenate the two arrays.
+				n := len(db.history.Logs.Chunk) - 1
+				copy(b, db.history.Logs.Chunk[:n])
+				if len(db.history.Logs.Chunk) > 2 {
+					// In case the array already contained log records, put the comma separator.
+					b[n] = ','
+					n++
+				}
+				copy(b[n:], buf[:last])
+				n += last
+				b[n] = ']'
+				n++
+
+				db.lock.Lock()
+				db.history.Logs.Chunk = b[:n]
+				db.lock.Unlock()
+
+				// Clear the valid/sent part of the buffer.
+				buf = buf[last+1:]
+			}
 		case errc := <-db.quit:
 			errc <- nil
 			return
-		case <-time.After(db.config.Refresh / 2):
-			db.sendToAll(&Message{
-				Logs: &LogsMessage{
-					Log: []string{fmt.Sprintf("%-4d: This is a fake log.", id)},
-				},
-			})
-			id++
 		}
 	}
 }
@@ -393,7 +563,7 @@ func (db *Dashboard) sendToAll(msg *Message) {
 	db.lock.Lock()
 	for _, c := range db.conns {
 		select {
-		case c.msg <- *msg:
+		case c.msg <- msg:
 		default:
 			c.conn.Close()
 		}

dashboard/message.go

@@ -16,7 +16,10 @@
 
 package dashboard
 
-import "time"
+import (
+	"encoding/json"
+	"time"
+)
 
 type Message struct {
 	General *GeneralMessage `json:"general,omitempty"`
@@ -28,18 +31,46 @@ type Message struct {
 	Logs    *LogsMessage    `json:"logs,omitempty"`
 }
 
+func (m *Message) DeepCopy() *Message {
+	return &Message{
+		m.General.DeepCopy(),
+		m.Home,
+		m.Chain,
+		m.TxPool,
+		m.Network,
+		m.System.DeepCopy(),
+		m.Logs,
+	}
+}
+
 type ChartEntries []*ChartEntry
 
+func (ce ChartEntries) DeepCopy() ChartEntries {
+	nce := make(ChartEntries, len(ce))
+	for i, v := range ce {
+		nce[i] = v.DeepCopy()
+	}
+	return nce
+}
+
 type ChartEntry struct {
 	Time  time.Time `json:"time,omitempty"`
 	Value float64   `json:"value,omitempty"`
 }
 
+func (ce *ChartEntry) DeepCopy() *ChartEntry {
+	return &ChartEntry{ce.Time, ce.Value}
+}
+
 type GeneralMessage struct {
 	Version string `json:"version,omitempty"`
 	Commit  string `json:"commit,omitempty"`
 }
 
+func (m *GeneralMessage) DeepCopy() *GeneralMessage {
+	return &GeneralMessage{m.Version, m.Commit}
+}
+
 type HomeMessage struct {
 	/* TODO (kurkomisi) */
 }
@@ -67,6 +98,27 @@ type SystemMessage struct {
 	DiskWrite      ChartEntries `json:"diskWrite,omitempty"`
 }
 
-type LogsMessage struct {
-	Log []string `json:"log,omitempty"`
+func (m *SystemMessage) DeepCopy() *SystemMessage {
+	return &SystemMessage{
+		m.ActiveMemory.DeepCopy(),
+		m.VirtualMemory.DeepCopy(),
+		m.NetworkIngress.DeepCopy(),
+		m.NetworkEgress.DeepCopy(),
+		m.ProcessCPU.DeepCopy(),
+		m.SystemCPU.DeepCopy(),
+		m.DiskRead.DeepCopy(),
+		m.DiskWrite.DeepCopy(),
+	}
+}
+
+type LogsMessage struct {
+	Chunk json.RawMessage `json:"chunk,omitempty"`
+}
+
+type Request struct {
+	Logs *LogsRequest `json:"logs,omitempty"`
+}
+
+type LogsRequest struct {
+	Time time.Time `json:"time,omitempty"`
 }

internal/debug/flags.go

@@ -95,7 +95,10 @@ var Flags = []cli.Flag{
 	memprofilerateFlag, blockprofilerateFlag, cpuprofileFlag, traceFlag,
 }
 
-var glogger *log.GlogHandler
+var (
+	ostream log.Handler
+	glogger *log.GlogHandler
+)
 
 func init() {
 	usecolor := term.IsTty(os.Stderr.Fd()) && os.Getenv("TERM") != "dumb"
@@ -103,22 +106,22 @@ func init() {
 	if usecolor {
 		output = colorable.NewColorableStderr()
 	}
-	glogger = log.NewGlogHandler(log.StreamHandler(output, log.TerminalFormat(usecolor)))
+	ostream = log.StreamHandler(output, log.TerminalFormat(usecolor))
+	glogger = log.NewGlogHandler(ostream)
 }
 
 // Setup initializes profiling and logging based on the CLI flags.
 // It should be called as early as possible in the program.
-func Setup(ctx *cli.Context, dashboard bool, path string) error {
+func Setup(ctx *cli.Context, disklogs string) error {
 	// logging
 	log.PrintOrigins(ctx.GlobalBool(debugFlag.Name))
+	if disklogs != "" {
+		glogger.SetHandler(log.MultiHandler(ostream, log.RotatingFileHandler(disklogs, 262144)))
+	}
 	glogger.Verbosity(log.Lvl(ctx.GlobalInt(verbosityFlag.Name)))
 	glogger.Vmodule(ctx.GlobalString(vmoduleFlag.Name))
 	glogger.BacktraceAt(ctx.GlobalString(backtraceAtFlag.Name))
-	h := log.Handler(glogger)
-	if dashboard {
-		h = log.MultiHandler(h, log.DashboardHandler(path+"/dashboard/logs"))
-	}
-	log.Root().SetHandler(h)
+	log.Root().SetHandler(glogger)
 
 	// profiling, tracing
 	runtime.MemProfileRate = ctx.GlobalInt(memprofilerateFlag.Name)

log/format.go

@@ -77,11 +77,11 @@ type TerminalStringer interface {
 // a terminal with color-coded level output and terser human friendly timestamp.
 // This format should only be used for interactive programs or while developing.
 //
-//     [TIME] [LEVEL] MESAGE key=value key=value ...
+//     [LEVEL] [TIME] MESAGE key=value key=value ...
 //
 // Example:
 //
-//     [May 16 20:58:45] [DBUG] remove route ns=haproxy addr=127.0.0.1:50002
+//     [DBUG] [May 16 20:58:45] remove route ns=haproxy addr=127.0.0.1:50002
 //
 func TerminalFormat(usecolor bool) Format {
 	return FormatFunc(func(r *Record) []byte {
@@ -202,6 +202,51 @@ func JSONFormat() Format {
 	return JSONFormatEx(false, true)
 }
 
+// JsonFormatOrderedCtx is similar to JsonFormatEx, except instead of a map this creates an array out of the ctx
+// in order to keep the original order.
+func JsonFormatOrderedCtx(pretty, lineSeparated bool) Format {
+	jsonMarshal := json.Marshal
+	if pretty {
+		jsonMarshal = func(v interface{}) ([]byte, error) {
+			return json.MarshalIndent(v, "", "    ")
+		}
+	}
+
+	return FormatFunc(func(r *Record) []byte {
+		props := make(map[string]interface{})
+
+		props[r.KeyNames.Time] = r.Time
+		props[r.KeyNames.Lvl] = r.Lvl.String()
+		props[r.KeyNames.Msg] = r.Msg
+
+		ctx := make([]string, len(r.Ctx))
+		for i := 0; i < len(r.Ctx); i += 2 {
+			k, ok := r.Ctx[i].(string)
+			if !ok {
+				props[errorKey] = fmt.Sprintf("%+v is not a string key,", r.Ctx[i])
+			}
+			ctx[i] = k
+			// TODO (kurkomisi): display hash fields entirely - possibly logger independently
+			ctx[i+1] = formatLogfmtValue(r.Ctx[i+1], true)
+		}
+		props[r.KeyNames.Ctx] = ctx
+
+		b, err := jsonMarshal(props)
+		if err != nil {
+			b, _ = jsonMarshal(map[string]string{
+				errorKey: err.Error(),
+			})
+			return b
+		}
+
+		if lineSeparated {
+			b = append(b, '\n')
+		}
+
+		return b
+	})
+}
+
 // JSONFormatEx formats log records as JSON objects. If pretty is true,
 // records will be pretty-printed. If lineSeparated is true, records
 // will be logged with a new line between each record.

log/handler.go

@@ -9,6 +9,9 @@ import (
 	"sync"
 
 	"github.com/go-stack/stack"
+	"io/ioutil"
+	"path/filepath"
+	"regexp"
 	"strings"
 )
 
@@ -71,32 +74,100 @@ func FileHandler(path string, fmtr Format) (Handler, error) {
 	return closingHandler{f, StreamHandler(f, fmtr)}, nil
 }
 
-// DashboardHandler returns a handler which writes log records to file chunks
-// at the given path. When a file's size reaches the 1MB, the handler creates
-// a new file named with the timestamp of the first log record it will contain.
-func DashboardHandler(path string) Handler {
-	if _, err := os.Stat(path); os.IsNotExist(err) {
-		if err = os.MkdirAll(path, 0755); err != nil {
-			// TODO (kurkomisi): handle error?
-			return DiscardHandler()
-		}
+type writeCounter struct {
+	w     io.Writer
+	count uint
 }
 
-	var size uint
-	maxSize := uint(1048576)
-	var h Handler
-	formatter := JsonFormat()
+func (w *writeCounter) Write(p []byte) (n int, err error) {
+	n, err = w.w.Write(p)
+	w.count += uint(n)
+	return n, err
+}
+
+// RotatingFileHandler returns a handler which writes log records to file chunks
+// at the given path. When a file's size reaches the limit, the handler creates
+// a new file named after the timestamp of the first log record it will contain.
+func RotatingFileHandler(path string, limit uint) Handler {
+	if err := os.MkdirAll(path, 0755); err != nil {
+		panic(err)
+	}
+	fmt.Println(path)
+	files, err := ioutil.ReadDir(path)
+	if err != nil {
+		panic(err)
+	}
+	counter := new(writeCounter)
+	h := StreamHandler(counter, JsonFormatOrderedCtx(false, true))
+
+	re := regexp.MustCompile(".log$")
+	var i int
+	for i = len(files) - 1; i >= 0 && (!files[i].Mode().IsRegular() || !re.Match([]byte(files[i].Name()))); i-- {
+	}
+	if i >= 0 {
+		// Open the last file, and continue to write into it until it's size reaches the limit.
+		last := files[i]
+		if last.Size() >= int64(limit) {
+			goto createNew
+		}
+		f, err := os.OpenFile(filepath.Join(path, last.Name()), os.O_RDWR|os.O_APPEND, 0644)
+		if err != nil {
+			goto createNew
+		}
+		// The previous execution could have been finished by interruption, in this case cut the invalid
+		// record from the end. Assume that every line ended by '\n' contains a valid log record.
+		bufSize := int64(100)
+		buf := make([]byte, bufSize)
+		cut := bufSize
+		if _, err = f.Seek(-bufSize, 2); err != nil {
+			goto createNew
+		}
+		n, err := f.Read(buf)
+		for err == nil {
+			for ; n > 0 && buf[n-1] != '\n'; n-- {
+			}
+			if n > 0 {
+				break
+			}
+			if _, err = f.Seek(-2*bufSize, 1); err != nil {
+				break
+			}
+			cut += bufSize
+			n, err = f.Read(buf)
+		}
+		if err != nil {
+			goto createNew
+		}
+		cut -= int64(n)
+
+		ns := last.Size() - cut
+		if err = f.Truncate(ns); err != nil {
+			goto createNew
+		}
+		counter.w = f
+		counter.count = uint(ns)
+	}
+createNew:
 
 	return FuncHandler(func(r *Record) error {
-		var err error
-		if h == nil || size > maxSize {
-			if h, err = FileHandler(fmt.Sprintf("%s/%s.log", path,
-				strings.Replace(r.Time.Format("060102150405.00"), ".", "", 1)), formatter); err != nil {
+		if counter.count > limit || counter.w == nil {
+			// TODO (kurkomisi): close the last file too.
+			if f, ok := counter.w.(*os.File); ok {
+				if err := f.Close(); err != nil {
 					return err
 				}
-			size = 0
 			}
-		size += uint(len(formatter.Format(r)))
+			f, err := os.OpenFile(
+				filepath.Join(path, fmt.Sprintf("%s.log", strings.Replace(r.Time.Format("060102150405.00"), ".", "", 1))),
+				os.O_CREATE|os.O_APPEND|os.O_WRONLY,
+				0644,
+			)
+			if err != nil {
+				return err
+			}
+			counter.w = f
+			counter.count = 0
+		}
 		return h.Log(r)
 	})
 }

log/handler_glog.go

@@ -57,6 +57,11 @@ func NewGlogHandler(h Handler) *GlogHandler {
 	}
 }
 
+// SetHandler updates the handler to write records to the specified sub-handler.
+func (h *GlogHandler) SetHandler(nh Handler) {
+	h.origin = nh
+}
+
 // pattern contains a filter for the Vmodule option, holding a verbosity level
 // and a file pattern to match.
 type pattern struct {

log/logger.go

@@ -11,6 +11,7 @@ import (
 const timeKey = "t"
 const lvlKey = "lvl"
 const msgKey = "msg"
+const ctxKey = "ctx"
 const errorKey = "LOG15_ERROR"
 
 type Lvl int
@@ -100,6 +101,7 @@ type RecordKeyNames struct {
 	Time string
 	Msg  string
 	Lvl  string
+	Ctx  string
 }
 
 // A Logger writes key/value pairs to a Handler
@@ -138,6 +140,7 @@ func (l *logger) write(msg string, lvl Lvl, ctx []interface{}) {
 			Time: timeKey,
 			Msg:  msgKey,
 			Lvl:  lvlKey,
+			Ctx:  ctxKey,
 		},
 	})
 }

node/config.go

@@ -179,7 +179,7 @@ func (c *Config) NodeDB() string {
 	if c.DataDir == "" {
 		return "" // ephemeral
 	}
-	return c.resolvePath(datadirNodeDatabase)
+	return c.ResolvePath(datadirNodeDatabase)
 }
 
 // DefaultIPCEndpoint returns the IPC path used by default.
@@ -262,8 +262,8 @@ var isOldGethResource = map[string]bool{
 	"trusted-nodes.json": true,
 }
 
-// resolvePath resolves path in the instance directory.
-func (c *Config) resolvePath(path string) string {
+// ResolvePath resolves path in the instance directory.
+func (c *Config) ResolvePath(path string) string {
 	if filepath.IsAbs(path) {
 		return path
 	}
@@ -309,7 +309,7 @@ func (c *Config) NodeKey() *ecdsa.PrivateKey {
 		return key
 	}
 
-	keyfile := c.resolvePath(datadirPrivateKey)
+	keyfile := c.ResolvePath(datadirPrivateKey)
 	if key, err := crypto.LoadECDSA(keyfile); err == nil {
 		return key
 	}
@@ -332,12 +332,12 @@ func (c *Config) NodeKey() *ecdsa.PrivateKey {
 
 // StaticNodes returns a list of node enode URLs configured as static nodes.
 func (c *Config) StaticNodes() []*discover.Node {
-	return c.parsePersistentNodes(c.resolvePath(datadirStaticNodes))
+	return c.parsePersistentNodes(c.ResolvePath(datadirStaticNodes))
 }
 
 // TrustedNodes returns a list of node enode URLs configured as trusted nodes.
 func (c *Config) TrustedNodes() []*discover.Node {
-	return c.parsePersistentNodes(c.resolvePath(datadirTrustedNodes))
+	return c.parsePersistentNodes(c.ResolvePath(datadirTrustedNodes))
 }
 
 // parsePersistentNodes parses a list of discovery node URLs loaded from a .json

node/node.go

@@ -570,12 +570,12 @@ func (n *Node) OpenDatabase(name string, cache, handles int) (ethdb.Database, er
 	if n.config.DataDir == "" {
 		return ethdb.NewMemDatabase(), nil
 	}
-	return ethdb.NewLDBDatabase(n.config.resolvePath(name), cache, handles)
+	return ethdb.NewLDBDatabase(n.config.ResolvePath(name), cache, handles)
 }
 
 // ResolvePath returns the absolute path of a resource in the instance directory.
 func (n *Node) ResolvePath(x string) string {
-	return n.config.resolvePath(x)
+	return n.config.ResolvePath(x)
 }
 
 // apis returns the collection of RPC descriptors this node offers.

node/service.go

@@ -43,7 +43,7 @@ func (ctx *ServiceContext) OpenDatabase(name string, cache int, handles int) (et
 	if ctx.config.DataDir == "" {
 		return ethdb.NewMemDatabase(), nil
 	}
-	db, err := ethdb.NewLDBDatabase(ctx.config.resolvePath(name), cache, handles)
+	db, err := ethdb.NewLDBDatabase(ctx.config.ResolvePath(name), cache, handles)
 	if err != nil {
 		return nil, err
 	}
@@ -54,7 +54,7 @@ func (ctx *ServiceContext) OpenDatabase(name string, cache int, handles int) (et
 // and if the user actually uses persistent storage. It will return an empty string
 // for emphemeral storage and the user's own input for absolute paths.
 func (ctx *ServiceContext) ResolvePath(path string) string {
-	return ctx.config.resolvePath(path)
+	return ctx.config.ResolvePath(path)
 }
 
 // Service retrieves a currently running service registered of a specific type.